Realtime Compute for Apache Flink：SQL常見問題

問題描述

當使用POJO類作為UDTF傳回型別，並在SQL中顯式聲明了UDTF返回列的別名列表（Alias Name）時，可能會出現欄位錯位（即使類型一致，但實際使用的欄位可能與預期不符）問題。

例如，如果使用如下POJO類作為UDTF的傳回型別，並根據自訂函數開發的要求進行打包並完成函數註冊（這裡使用作業級自訂函數註冊方式）後，SQL校正會失敗。

package com.aliyun.example;

public class TestPojoWithoutConstructor {
	public int c;
	public String d;
	public boolean a;
	public String b;
}

package com.aliyun.example;

import org.apache.flink.table.functions.TableFunction;

public class MyTableFuncPojoWithoutConstructor extends TableFunction<TestPojoWithoutConstructor> {
	private static final long serialVersionUID = 1L;

	public void eval(String str1, Integer i2) {
		TestPojoWithoutConstructor p = new TestPojoWithoutConstructor();
		p.d = str1 + "_d";
		p.c = i2 + 2;
		p.b = str1 + "_b";
		collect(p);
	}
}

CREATE TEMPORARY FUNCTION MyTableFuncPojoWithoutConstructor as 'com.aliyun.example.MyTableFuncPojoWithoutConstructor';

CREATE TEMPORARY TABLE src ( 
  id STRING,
  cnt INT
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE sink ( 
  f1 INT,
  f2 STRING,
  f3 BOOLEAN,
  f4 STRING
) WITH (
 'connector' = 'print'
);

INSERT INTO sink
SELECT T.* FROM src, LATERAL TABLE(MyTableFuncPojoWithoutConstructor(id, cnt)) AS T(c, d, a, b);

SQL校正報錯資訊如下：

org.apache.flink.table.api.ValidationException: SQL validation failed. Column types of query result and sink for 'vvp.default.sink' do not match.
Cause: Sink column 'f1' at position 0 is of type INT but expression in the query is of type BOOLEAN NOT NULL.
Hint: You will need to rewrite or cast the expression.

Query schema: [c: BOOLEAN NOT NULL, d: STRING, a: INT NOT NULL, b: STRING]
Sink schema:  [f1: INT, f2: STRING, f3: BOOLEAN, f4: STRING]
	at org.apache.flink.table.sqlserver.utils.FormatValidatorExceptionUtils.newValidationException(FormatValidatorExceptionUtils.java:41)

看起來從UDTF返回的欄位和POJO類中的欄位可能錯位了，SQL中欄位c最終是BOOLEAN，而欄位a是INT類型，和POJO類的定義恰好相反。

問題原因

根據POJO類的類型規則：

• 如果POJO類實現了有參建構函式，推導的傳回型別會按建構函式的參數列表順序。

• 如果POJO類缺少有參建構函式，就會按欄位名的字典序重排列。

在上述樣本中，由於UDTF傳回型別缺少有參建構函式，因此對應的傳回型別為BOOLEAN a, VARCHAR(2147483647) b, INTEGER c, VARCHAR(2147483647) d)。雖然這一步並沒有產生錯誤，但因為SQL中對返回欄位加了重新命名列表LATERAL TABLE(MyTableFuncPojoWithoutConstructor(id, cnt)) AS T(c, d, a, b)，這導致對推匯出的類型顯式進行了重新命名（基於欄位位置進行映射），進而引發與POJO類中的欄位錯位問題，出現校正異常或非預期的資料錯位問題。

解決方案

• POJO類缺少有參建構函式時，去掉對UDTF返回欄位的顯式重新命名，如將上述SQL的INSERT語句改為：

  -- POJO類無有參建構函式時，推薦明確選取需要的欄位名，使用 T.* 時需要明確知曉實際返回的欄位順序。
  SELECT T.c, T.d, T.a, T.b FROM src, LATERAL TABLE(MyTableFuncPojoWithoutConstructor(id, cnt)) AS T;

• POJO類實現有參建構函式，以確定傳回型別的欄位順序。這種情況下UDTF傳回型別的欄位順序就是有參建構函式的參數順序。

  package com.aliyun.example;
  
  public class TestPojoWithConstructor {
  	public int c;
  	public String d;
  	public boolean a;
  	public String b;
  
  	// Using specific fields order instead of alphabetical order
  	public TestPojoWithConstructor(int c, String d, boolean a, String b) {
  		this.c = c;
  		this.d = d;
  		this.a = a;
  		this.b = b;
  	}
  }

問題描述

如果作業未設定table.exec.mini-batch.size或設定table.exec.mini-batch.size為負，並且作業還包含WindowAggregate和GroupAggregate，且WindowAggregate的時間列為事件時間（proctime），在這種情況下啟動作業，在作業拓撲圖中會看到包含LocalGroupAggregate節點，但缺失MiniBatchAssigner節點。

作業中同時包含WindowAggregate和GroupAggregate，且WindowAggregate的時間列為事件時間（proctime）的程式碼範例如下。

CREATE TEMPORARY TABLE s1 (
  a INT,
  b INT,
  ts as PROCTIME(),
  PRIMARY KEY (a) NOT ENFORCED
) WITH (
  'connector'='datagen',
  'rows-per-second'='1',
  'fields.b.kind'='random',
  'fields.b.min'='0',
  'fields.b.max'='10'
);

CREATE TEMPORARY TABLE sink (
  a BIGINT,
  b BIGINT
) WITH (
  'connector'='print'
);

CREATE TEMPORARY VIEW window_view AS
SELECT window_start, window_end, a, sum(b) as b_sum FROM TABLE(TUMBLE(TABLE s1, DESCRIPTOR(ts), INTERVAL '2' SECONDS)) GROUP BY window_start, window_end, a;

INSERT INTO sink SELECT count(distinct a), b_sum FROM window_view GROUP BY b_sum;

問題原因

在table.exec.mini-batch.size未配置或者為負的情況下，MiniBatch處理模式會使用Managed Memory快取資料，但是此時會錯誤地無法產生MiniBatchAssigner節點，因此計算節點無法收到MinibatchAssigner節點發送的Watermark訊息後觸發計算和輸出，只能在三種條件（Managed Memory已滿、進行Checkpoint前和作業停止時）之一下觸發計算和輸出，詳情請參見table.exec.mini-batch.size。如果此時Checkpoint間隔設定過大，就會導致資料積攢在LocalGroupAggregate節點中，長時間無法輸出。

解決方案

• 調小Checkpoint間隔，讓LocalGroupAggregate節點在執行Checkpoint前自動觸發輸出。

• 通過Heap Memory來快取資料，讓LocalGroupAggregate節點內快取資料達到N條時自動觸發輸出。即在營運中心 > 作業營運頁面的部署詳情頁運行參數配置地區其他配置中，設定table.exec.mini-batch.size參數為正值N。

原因1：聲明源表Watermark時使用了TIMESTAMP_LTZ（TIMESTAMP(p) WITH LOCAL TIME ZONE）類型，導致Watermark和目前時間有時差。

下文以具體的樣本為您展示使用TIMESTAMP_LTZ類型和TIMESTAMP類型對應的Watermark指標差異。

• 源表中Watermark聲明使用的欄位是TIMESTAMP_LTZ類型。

  CREATE TEMPORARY TABLE s1 (
    a INT,
    b INT,
    ts as CURRENT_TIMESTAMP,--使用CURRENT_TIMESTAMP內建函數產生TIMESTAMP_LTZ類型。
    WATERMARK FOR ts AS ts - INTERVAL '5' SECOND 
  ) WITH (
    'connector'='datagen',
    'rows-per-second'='1',
    'fields.b.kind'='random','fields.b.min'='0','fields.b.max'='10'
  );
  
  CREATE TEMPORARY TABLE t1 (
    k INT,
    ts_ltz timestamp_ltz(3),
    cnt BIGINT
  ) WITH ('connector' = 'print');
  
  -- 輸出計算結果。
  INSERT INTO t1
  SELECT b, window_start, COUNT(*) FROM
  TABLE(
      TUMBLE(TABLE s1, DESCRIPTOR(ts), INTERVAL '5' SECOND))
  GROUP BY b, window_start, window_end;

  說明

  Legacy Window對應的老文法和TVF Window（Table-Valued Function）產生的結果是一致的。以下為Legacy Window對應的老文法的範例程式碼。

  SELECT b, TUMBLE_END(ts, INTERVAL '5' SECOND), COUNT(*) FROM s1 GROUP BY TUMBLE(ts, INTERVAL '5' SECOND), b;

  在Flink開發控制台將作業部署上線運行後，以北京時間為例，可以觀察到作業運行拓撲圖及監控警示上顯示的Watermark和目前時間存在8小時時差。

  • Watermark&Low Watermark

    

  • Task InputWatermark

    

• 源表中Watermark聲明使用的欄位是TIMESTAMP（TIMESTAMP(p) WITHOUT TIME ZONE）類型。

  CREATE TEMPORARY TABLE s1 (
    a INT,
    b INT,
    -- 類比資料源中的TIMESTAMP無時區資訊，從2024-01-31 01:00:00開始逐秒累加。
    ts as TIMESTAMPADD(SECOND, a, TIMESTAMP '2024-01-31 01:00:00'),
    WATERMARK FOR ts AS ts - INTERVAL '5' SECOND 
  ) WITH (
    'connector'='datagen',
    'rows-per-second'='1',
    'fields.a.kind'='sequence','fields.a.start'='0','fields.a.end'='100000',
    'fields.b.kind'='random','fields.b.min'='0','fields.b.max'='10'
  );
  
  CREATE TEMPORARY TABLE t1 (
    k INT,
    ts_ltz timestamp_ltz(3),
    cnt BIGINT
  ) WITH ('connector' = 'print');
  
  -- 輸出計算結果。
  INSERT INTO t1
  SELECT b, window_start, COUNT(*) FROM
  TABLE(
      TUMBLE(TABLE s1, DESCRIPTOR(ts), INTERVAL '5' SECOND))
  GROUP BY b, window_start, window_end;

  在Flink開發控制台上將作業部署上線運行後，可以觀察到作業運行拓撲圖及監控警示上顯示的Watermark和目前時間是同步的（本樣本是與類比資料的時間同步的），不存在時差現象。

  • Watermark&Low Watermark

    

  • Task InputWatermark

    

原因2：Flink開發控制台和Apache Flink UI的展示時間存在時區差異。

Flink開發控制台UI介面是以UTC+0顯示時間，而Apache Flink UI是通過瀏覽器擷取本地時區並進行相應的時間轉換後的本地時間。以北京時間為例，為您展示二者顯示區別，您會觀察到在Flink開發控制台顯示的時間比Apache Flink UI時間慢8小時。