Flink 1.17.1有界流水位线注入数量及调试打印方法咨询

原始代码 #

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

SingleOutputStreamOperator<Tuple2<String, Integer>> dataStream = env.fromElements(
    Tuple2.of("01", 1),
    Tuple2.of("02", 2),
    Tuple2.of("03", 3),
    Tuple2.of("04", 4),
    Tuple2.of("05", 5)
).assignTimestampsAndWatermarks(
    WatermarkStrategy.<Tuple2<String, Integer>>forMonotonousTimestamps()
        .withTimestampAssigner(
            (tuple, ts) -> System.currentTimeMillis())
);
env.execute("dfjghf");

env.execute("gghh");

技术问题及解答 #

1. 该有界流中会注入多少条水位线？是否为5条？ #

不是5条，实际只会注入1条水位线。

原因：使用forMonotonousTimestamps()的水位线策略时，Flink对于有界流不会为每个数据元素生成水位线。该策略基于单调递增时间戳更新水位线，但在有界流场景下，Flink会在所有数据处理完成后，发送一条时间戳为Long.MAX_VALUE的最终水位线，以此标识流结束，不会为每个元素单独生成水位线。

2. 如何打印这些水位线元素以用于调试？ #

有两种实用方式：

方式一：通过日志打印

修改Flink的日志配置（如log4j2.xml或logback.xml），将org.apache.flink.streaming.api.watermark的日志级别设为DEBUG，Flink会自动在日志中输出水位线的生成和传递信息。示例log4j2配置片段：

<Logger name="org.apache.flink.streaming.api.watermark" level="DEBUG" additivity="false">
    <AppenderRef ref="ConsoleAppender"/>
</Logger>

方式二：自定义ProcessFunction捕获水位线

在数据流中插入ProcessFunction，重写processWatermark方法打印水位线，同时保证水位线正常传递到下游。示例代码：

dataStream.process(new ProcessFunction<Tuple2<String, Integer>, Tuple2<String, Integer>>() {
    @Override
    public void processElement(Tuple2<String, Integer> value, Context ctx, Collector<Tuple2<String, Integer>> out) throws Exception {
        // 正常传递数据元素
        out.collect(value);
    }

    @Override
    public void processWatermark(Watermark mark, Context ctx, Collector<Tuple2<String, Integer>> out) throws Exception {
        // 打印水位线信息
        System.out.println("捕获到水位线：" + mark.getTimestamp());
        // 手动传递水位线，避免下游算子无法接收
        ctx.emitWatermark(mark);
    }
}).print();

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