Flink Table API滚动窗口聚合后Kafka sink主题无输出问题求助

问题根因排查及修复方案 #

1. 核心错误：事件时间字段硬编码 #

当前transaction_time字段定义存在逻辑错误：你配置的transaction_time AS TO_TIMESTAMP_LTZ(4001, 3)生成的是固定时间戳（对应1970-01-01 00:00:04.001 UTC），所有数据的事件时间完全固定，导致水位线永远无法推进到2分钟窗口的结束阈值，窗口永远不会触发计算输出。

修复方案：

将事件时间绑定到实际业务时间字段，比如你已经通过元数据获取到Debezium的事件生成时间，直接复用该字段即可：

-- 修改源表的时间和水位线定义，删除硬编码的transaction_time
CREATE TABLE transactions (
  event_time TIMESTAMP(3) METADATA FROM 'value.source.timestamp' VIRTUAL,
  id INT PRIMARY KEY,
  transaction_status STRING,
  transaction_type STRING,
  merchant_id INT,
  -- 直接基于event_time定义水位线
  WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
) WITH (
  -- 原有WITH参数保持不变
  'debezium-json.schema-include' = 'true' ,
  'connector' = 'kafka',
  'topic' = 'dbserver1.inventory.transactions',
  'properties.bootstrap.servers' = 'my-cluster-kafka-bootstrap.kafka.svc:9092',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'debezium-json'
)

同步修改窗口逻辑中的事件时间字段：

public static Table report(Table transactions) {
    return transactions
            // 将transaction_time替换为event_time
            .window(Tumble.over(lit(2).minutes()).on($("event_time")).as("w"))
            .groupBy($("w"), $("transaction_status"))
            .select(
                    $("w").start().as("window_start"),
                    $("w").end().as("window_end"),
                    $("transaction_status"),
                    $("id").count().as("id_count"));
}

如果需要使用业务数据里的事务发生时间，从Debezium的payload中读取对应字段转成TIMESTAMP类型即可，不要使用固定值。

2. 并行度水位线对齐问题 #

作业并行度设置为2时，需要满足两个条件才能正常推进全局水位线：

• 源Kafka主题的分区数≥2，否则空闲的并行子任务水位线会一直停留在初始值，拉低全局水位线
• 如果存在低流量/空闲分区，需要开启空闲分区检测，在源表WITH参数中添加：

  'table.exec.source.idle-timeout' = '10s'
  

  表示10秒无数据的分区会被标记为空闲，不参与全局水位线计算。

3. Sink主键配置错误 #

当前upsert-kafka的主键仅配置了window_start，但你的聚合维度是窗口+transaction_status，同一个窗口内会有多条不同transaction_status的结果，单主键会导致数据被覆盖。修改sink表的主键定义：

CREATE TABLE my_report (
  window_start TIMESTAMP(3),
  window_end TIMESTAMP(3),
  transaction_status STRING,
  id_count BIGINT,
  -- 主键改为窗口+维度的联合主键
  PRIMARY KEY (window_start, window_end, transaction_status) NOT ENFORCED
) WITH (
  -- 原有WITH参数保持不变
  'connector' = 'upsert-kafka',
  'topic' = 'dbserver1.inventory.my-window-sink',
  'properties.bootstrap.servers' = 'my-cluster-kafka-bootstrap.kafka.svc:9092',
  'properties.group.id' = 'testGroup',
  'key.format' = 'json',
  'value.format' = 'json'
)

验证方法 #

修复完成后可以先将sink临时改为print连接器，确认窗口有输出后再对接Kafka，也可以在Flink UI的任务监控页查看各子任务的水位线数值，确认水位线持续推进即可。

内容的提问来源于stack exchange，提问作者Nada Makram