問題描述

當使用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小時。