实时计算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小时。