Flink基于ProtoBuf自定义类配置WatermarkStrategy实现滚动窗口聚合的问题及TIMESTAMP精度排查

你遇到的核心问题是：虽然你在Stream API里分配了水位线，但Flink SQL并不知道ts列是事件时间属性（Event Time Attribute）——它只是一个普通的TIMESTAMP(9)列，而窗口聚合必须基于标记为事件时间的列才能执行。

另外，Java的LocalDateTime和java.sql.Timestamp默认会被Flink解析为纳秒精度的TIMESTAMP(9)，这也是为什么你调整精度后依然没效果的原因。下面给你两种可行的解决方案：

方案一：在Table层面转换并标记事件时间属性 #

这种方式不需要在Stream层提前转成Timestamp，而是保留毫秒级的epoch时间戳，在创建Table视图时通过SQL函数转换并标记为事件时间：

// 1. 从Proto对象提取所需字段，保留毫秒级时间戳为Long类型
DataStream<Row> mappedStream = stream
    .returns(BiddingEvent.BidEvent.class)
    .map(e -> Row.of(
        e.getTracking().getCampaign().getId(),
        e.getTracking().getAuction().getId(),
        e.getTimestamp().getMilliseconds() // 保留原始毫秒值
    ))
    .returns(Types.ROW_NAMED(
        new String[] {"campaign_id", "auction_id", "ts_ms"},
        Types.STRING, Types.STRING, Types.LONG
    ));

// 2. 创建Table并显式定义事件时间列和水位线
Table bideventsTable = tableEnv.fromDataStream(
    mappedStream,
    $("campaign_id"),
    $("auction_id"),
    $("ts_ms"),
    // 将毫秒级epoch转成TIMESTAMP(3)，并标记为事件时间列
    $("ts").expr("TO_TIMESTAMP_LTZ(ts_ms, 3)").rowtime(),
    // 定义水位线（和你之前的10秒乱序窗口一致）
    $("watermark").expr("ts - INTERVAL '10' SECOND").watermark()
);

// 3. 注册为临时视图
tableEnv.createTemporaryView("bidevents", bideventsTable);

方案二：在Stream层分配水位线后，关联到Table的事件时间列 #

如果你更倾向于在Stream层完成水位线分配，可以这样做：

// 1. 先映射出包含毫秒时间戳的Row
DataStream<Row> mappedStream = stream
    .returns(BiddingEvent.BidEvent.class)
    .map(e -> Row.of(
        e.getTracking().getCampaign().getId(),
        e.getTracking().getAuction().getId(),
        e.getTimestamp().getMilliseconds()
    ))
    .returns(Types.ROW_NAMED(
        new String[] {"campaign_id", "auction_id", "ts_ms"},
        Types.STRING, Types.STRING, Types.LONG
    ));

// 2. 在Stream层分配水位线（基于毫秒时间戳）
WatermarkStrategy<Row> watermarkStrategy = WatermarkStrategy
    .<Row>forBoundedOutOfOrderness(Duration.ofSeconds(10))
    .withTimestampAssigner((event, ts) -> (Long) event.getField("ts_ms"));

DataStream<Row> streamWithWatermark = mappedStream.assignTimestampsAndWatermarks(watermarkStrategy);

// 3. 转成Table时，显式将毫秒时间戳转成TIMESTAMP(3)并标记为事件时间
Table bideventsTable = tableEnv.fromDataStream(
    streamWithWatermark,
    Schema.newBuilder()
        .column("campaign_id", "STRING")
        .column("auction_id", "STRING")
        .column("ts_ms", "BIGINT")
        // 转换为毫秒精度的TIMESTAMP
        .column("ts", "TIMESTAMP(3)")
        .computedColumn("ts", "TO_TIMESTAMP_LTZ(ts_ms, 3)")
        // 关联Stream层的水位线策略
        .watermark("ts", "ts - INTERVAL '10' SECOND")
        .build()
);

tableEnv.createTemporaryView("bidevents", bideventsTable);

为什么之前的代码不行？ #

1. 你在map里把时间转成了Java的Timestamp/LocalDateTime，这些类型在Flink中默认对应TIMESTAMP(9)（纳秒精度），而Flink SQL的事件时间属性要求精度≤3（毫秒及以下）；
2. 你没有显式告诉Flink SQL这个ts列是事件时间属性——仅仅在Stream层分配水位线是不够的，必须在Table/Schema层面标记rowtime，让SQL引擎识别它是用于时间窗口的列。

完成上述修改后，你再执行DESCRIBE bidevents，就能看到ts列的类型是TIMESTAMP(3) *ROWTIME*，这时候就可以正常执行TUMBLE窗口聚合查询了。

内容的提问来源于stack exchange，提问作者Elmar Macek