0

Kafka服务器中有一个主题。在程序中，我们将此主题作为流读取并分配事件时间戳。然后对这个流做窗口操作。但该程序不起作用。调试后，似乎没有执行WindowOperator的processWatermark方法。这是我的代码。

    DataStream<Tuple2<String, Long>> advertisement = env
            .addSource(new FlinkKafkaConsumer082<String>("advertisement", new SimpleStringSchema(), properties))
            .map(new MapFunction<String, Tuple2<String, Long>>() {
                private static final long serialVersionUID = -6564495005753073342L;

                @Override
                public Tuple2<String, Long> map(String value) throws Exception {
                    String[] splits = value.split(" ");
                    return new Tuple2<String, Long>(splits[0], Long.parseLong(splits[1]));
                }
            }).assignTimestamps(timestampExtractor);

    advertisement
            .keyBy(keySelector)
            .window(TumblingTimeWindows.of(Time.of(10, TimeUnit.SECONDS)))
            .apply(new WindowFunction<Tuple2<String,Long>, Integer, String, TimeWindow>() {
                private static final long serialVersionUID = 5151607280638477891L;
                @Override
                public void apply(String s, TimeWindow window, Iterable<Tuple2<String, Long>> values, Collector<Integer> out) throws Exception {
                    out.collect(Iterables.size(values));
                }
            }).print();

为什么会这样？如果我在“assignTimestamps(timestampExtractor)”之前添加“keyBy(keySelector)”，那么程序就可以工作。任何人都可以帮助解释原因吗？

4

1 回答 1

2

您受到 Flink 中的一个已知错误的影响：FLINK-3121:Watermark forwarding does not work for sources not generate any data。

问题是有更多的 FlinkKafkaConsumer 正在运行（很可能是 CPU 核心的数量（比如 4））然后你有分区（1）。只有一个 Kafka 消费者在发射水印，其他消费者处于空闲状态。

窗口操作员没有意识到这一点，等待所有消费者的水印到达。这就是为什么窗户永远不会触发的原因。

于 2015-12-05T09:38:46.387 回答