加入流 Flink 不适用于 Kafka 消费者

Question

我正在尝试加入两个流，一个来自数据收集，一个来自 Kafka。

代码 sn-p

public static void main(String[] args) {
        KafkaSource<JsonNode> kafkaSource = ...
        
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // Kafka messages : {"name": "John"} 
        final DataStream<JsonNode> dataStream1 = env.fromSource(kafkaSource, waterMark(), "Kafka").rebalance()
                .assignTimestampsAndWatermarks(waterMark());
        
        final DataStream<String> dataStream2 = env.fromElements("John", "Zbe", "Abe")
                .assignTimestampsAndWatermarks(waterMark());
        
        dataStream1
            .join(dataStream2)
            .where(new KeySelector<JsonNode, String>() {
    
                @Override
                public String getKey(JsonNode value) throws Exception {
                    return value.get("name").asText();
                }
            })
            .equalTo(new KeySelector<String, String>() {

                @Override
                public String getKey(String value) throws Exception {
                    return value;
                }
            })
            .window(SlidingEventTimeWindows.of(Time.minutes(50) /* size */, Time.minutes(10) /* slide */))
            .apply(new JoinFunction<JsonNode, String, String>() {
    
                @Override
                public String join(JsonNode first, String second) throws Exception {

                    return first+" "+second;
                }
            }).print();
            
            env.execute();
    }

水印

private static <T>  WatermarkStrategy<T> waterMark() {
        return new WatermarkStrategy<T>() {

            @Override
            public WatermarkGenerator<T> createWatermarkGenerator(
                    org.apache.flink.api.common.eventtime.WatermarkGeneratorSupplier.Context context) {
                return new AscendingTimestampsWatermarks<>();
            }
            
            @Override
            public TimestampAssigner<T> createTimestampAssigner(TimestampAssignerSupplier.Context context) {
                return (event, timestamp) -> System.currentTimeMillis();
            }
            
        };
    }

运行 sn-p 代码后，输出中没有任何合并数据。我是不是哪里出错了？

Apache flink 版本：1.13.2

Answer 1

问题可能与水印有关。由于您没有使用基于事件时间的时间戳，请尝试将 SlidingEventTimeWindows 更改为 SlidingProcessingTimeWindows 并查看它是否会产生结果。