求助：添加Watermark与聚合后Flink任务无标准输出

问题描述 #

遇到Flink任务无法输出至标准输出的问题：移除Filter和Watermark后，能正常接收并打印Kafka主题的原始消息；但添加Watermark与窗口聚合逻辑后，却没有任何输出。以下是main方法中的代码：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// create kafka source
KafkaSource<VideoAdCompletedEvent> videoAdsSource = KafkaSource
    .builder()
    .setBootstrapServers(jobConfig.kafkaSourceServer)
    .setTopics(jobConfig.sourceKafkaTopic).setGroupId(jobConfig.consumerGroupId)
    .setStartingOffsets(OffsetsInitializer.earliest())
    .setValueOnlyDeserializer(new VideoDeserSchema())
    .build();

WatermarkStrategy<VideoAdCompletedEvent> watermarkStrategy = WatermarkStrategy
    .<VideoAdCompletedEvent>forBoundedOutOfOrderness(Duration.ofSeconds(30)) // allow events up to 30 seconds out of order
    .withTimestampAssigner(
        (event, timestamp) -> event.getOriginTs()
    )
    .withIdleness(Duration.ofMillis(500));

DataStream<VideoAdCompletedEvent> videoAdsStream = env
    .fromSource(videoAdsSource, watermarkStrategy, "Mobile Analytics Filtered Video Ads Topic")
    .assignTimestampsAndWatermarks(watermarkStrategy);

DataStream<VideoAdCompletedEvent> filteredStream = videoAdsStream
    .filter(new FilterFunction<VideoAdCompletedEvent>() {
        @Override
        public boolean filter(VideoAdCompletedEvent message) throws Exception {
          return (message.getCompletedReason().equals("VIDEO_FINISHED") || message.getCompletedReason().equals("video_finished"));
        }
    });

DataStream<Tuple2<String, Integer>>  aggStream = filteredStream
    .keyBy((VideoAdCompletedEvent event) -> event.getVideoId())
    .window(TumblingEventTimeWindows.of(Time.seconds(30)))
    .aggregate(new DistinctUserIdAggregate(), new VideoIdCountProcessFunction());

SinkFunction<VideoAdCompletedEvent> printSink = new PrintSinkFunction();

aggStream.addSink(printSink);

env.execute("Test Job");

可能的原因及解决办法 #

• 重复设置Watermark导致逻辑异常：代码中通过fromSource已经传入了watermarkStrategy，后续又调用assignTimestampsAndWatermarks(watermarkStrategy)重复配置，会覆盖之前的时间戳分配和Watermark生成逻辑，导致Watermark无法正常推进。解决：删除assignTimestampsAndWatermarks(watermarkStrategy)这一行，仅保留fromSource中的Watermark配置。
• 事件时间窗口未触发计算：事件时间窗口的触发依赖Watermark推进到窗口结束时间。如果event.getOriginTs()返回的时间戳不符合预期（比如是未来时间、或当前Watermark未达到窗口关闭时间点），窗口不会触发输出。解决：验证OriginTs是否为合法的毫秒级时间戳；通过Flink UI查看Watermark的推进情况；测试阶段可改用处理时间窗口（TumblingProcessingTimeWindows）快速验证聚合逻辑。
• Filter过滤了全部数据：检查Kafka消息的completedReason字段值，是否存在大小写不一致、拼写错误等情况，导致Filter条件全部不满足，filteredStream无数据流入后续窗口。解决：在Filter前后添加print()算子，查看经过过滤后的数据流是否有数据；或临时放宽Filter条件进行测试。
• Sink类型不兼容：aggStream的类型是Tuple2<String, Integer>，但创建的printSink是SinkFunction<VideoAdCompletedEvent>，类型不匹配会导致数据无法正常输出（可能编译无报错，但运行时数据被丢弃）。解决：将Sink改为匹配类型，比如使用new PrintSinkFunction<Tuple2<String, Integer>>()。
• 自定义聚合/窗口函数逻辑错误：DistinctUserIdAggregate或VideoIdCountProcessFunction的实现存在问题，比如累加器未正确统计数据、ProcessWindowFunction未调用collect输出结果。解决：检查这两个自定义函数的实现，确保AggregateFunction的add/getResult方法逻辑正确，ProcessWindowFunction的process方法正常输出结果。

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