ApacheFlink流连接使用数据流API，不输出任何内容

Question

我有2个使用kafka主题创建的流，我正在使用DataStream API加入它们。我希望将连接（应用）的结果发布到另一个kafka主题。我在外部主题中看不到连接的结果。

我确认我向两个源主题发布了正确的数据。不确定哪里出了问题。下面是代码片段，

创建的流如下所示。

DataStream<String> ms1=env.addSource(new FlinkKafkaConsumer("top1",new SimpleStringSchema(),prop))
            .assignTimestampsAndWatermarks(new WatermarkStrategy() {
                @Override
                public WatermarkGenerator createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
                    return new AscendingTimestampsWatermarks<>();
                }
                @Override
                public TimestampAssigner createTimestampAssigner(TimestampAssignerSupplier.Context context) {
                    return (event, timestamp) -> System.currentTimeMillis();
                }
            });
DataStream<String> ms2=env.addSource(new FlinkKafkaConsumer("top2",new SimpleStringSchema(),prop))
            .assignTimestampsAndWatermarks(new WatermarkStrategy() {
                @Override
                public WatermarkGenerator createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
                    return new AscendingTimestampsWatermarks<>();
                }
                @Override
                public TimestampAssigner createTimestampAssigner(TimestampAssignerSupplier.Context context) {
                    return (event, timestamp) -> System.currentTimeMillis();
                }
            });

流连接使用等于的连接执行，如下所示。

DataStream joinedStreams = ms1.join(ms2)
            .where(o -> {String[] tokens = ((String) o).split("::"); return tokens[0];})
            .equalTo(o -> {String[] tokens = ((String) o).split("::"); return tokens[0];})
            .window(EventTimeSessionWindows.withGap(Time.seconds(60)))
            .apply(new JoinFunction<String, String, CountryData>() {
                @Override
                public CountryData join(String o, String o2) throws Exception {
                    String[] tokens1 = o.split("::");
                    String[] tokens2 = o2.split("::");
                    CountryData countryData = new CountryData(tokens1[0], tokens1[1], tokens1[2], Long.parseLong(tokens1[3])+Long.parseLong(tokens2[3]));
                    return countryData;
                }});

如下所述，

joinedStreams.addSink(new FlinkKafkaProducer<CountryData>("localhost:9095","flink-output", new CustomSchema()));
dataStreamSink.setParallelism(1);
dataStreamSink.name("KAFKA-TOPIC");

有什么线索吗，哪里出了问题？我可以在拓扑中看到可用的消息，谢谢

Answer 1

我认为这两个＜code＞FlinkkaFaConsumer＜/code＞实例缺少时间提取器和水印配置。

由于代码使用的是事件-时间窗口连接，因此它需要与在Kafka中找到的数据相关联的某种时间信息，以便知道每个事件对应于哪个时间窗口。

如果没有这一点，两个流中的事件在事件时间上可能永远不会足够接近EventTimeSessionWindows.withGap（time.seconds（60））定义的60秒窗口。

您还需要设置水印参数，以告诉 Flink 何时停止等待新数据并具体化输出 s.t.，您可以看到连接结果。

查看Kafka连接器时间和水印配置，了解您拥有的各种时间提取和水印可能性。

最后，确保将足够长时间内的测试数据发送到您的应用程序中。对于事件时间处理，只有“足够老”的数据才能输出，年轻的数据总是“在传输中被卡住”。例如，对于60秒的时间窗口和30秒的水位线，您至少需要90秒的数据才能在输出中看到任何内容。