Re: Kafka 数据源无法实现基于事件时间的窗口聚合

[Weihua Hu]

Hi,

问题应该是 kafka source 配置了多并发运行，但数据量比较少（或者 topic 的 partition 数量小于 task
的并发数量），不是所有的 source task 都消费到了数据并产生 watermark，导致下游聚合算子无法对齐 watermark 触发计算。
可以尝试通过以下办法解决：
1. 将 source 并发控制为 1
2. 为 watermark 策略开始 idleness 处理，参考 [#1]

fromElement 数据源会强制指定并发为 1

[#1]
https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/event-time/generating_watermarks/#dealing-with-idle-sources


Best,
Weihua


On Tue, Feb 7, 2023 at 1:31 PM wei_yuze <wei_y...@qq.com.invalid> wrote:

> 您好！
>
>
>
>
> 我在进行基于事件时间的窗口聚合操作时，使用fromElement数据源可以实现，但替换为Kafka数据源就不行了，但程序并不报错。以下贴出代码。代码中给了两个数据源，分别命名为：streamSource
> 和 kafkaSource
> 。当使用streamSource生成watermarkedStream的时候，可以完成聚合计算并输出结果。但使用kafkaSource却不行。
>
>
>
>
> public class WindowReduceTest2 {&nbsp; &nbsp; public static void
> main(String[] args) throws Exception {
> &nbsp; &nbsp; &nbsp; &nbsp; StreamExecutionEnvironment env =
> StreamExecutionEnvironment.getExecutionEnvironment();
>
>
> &nbsp; &nbsp; &nbsp; &nbsp; // 使用fromElement数据源
> &nbsp; &nbsp; &nbsp; &nbsp; DataStreamSource<Event2&gt; streamSource =
> env.fromElements(
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; new
> Event2("Alice", "./home", "2023-02-04 17:10:11"),
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; new Event2("Bob",
> "./cart", "2023-02-04 17:10:12"),
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; new
> Event2("Alice", "./home", "2023-02-04 17:10:13"),
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; new
> Event2("Alice", "./home", "2023-02-04 17:10:15"),
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; new Event2("Cary",
> "./home", "2023-02-04 17:10:16"),
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; new Event2("Cary",
> "./home", "2023-02-04 17:10:16")
> &nbsp; &nbsp; &nbsp; &nbsp; );
>
>
> &nbsp; &nbsp; &nbsp; &nbsp; // 使用Kafka数据源
> &nbsp; &nbsp; &nbsp; &nbsp; JsonDeserializationSchema<Event2&gt;
> jsonFormat = new JsonDeserializationSchema<&gt;(Event2.class);
> &nbsp; &nbsp; &nbsp; &nbsp; KafkaSource<Event2&gt; source =
> KafkaSource.<Event2&gt;builder()
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
> .setBootstrapServers(Config.KAFKA_BROKERS)
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
> .setTopics(Config.KAFKA_TOPIC)
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
> .setGroupId("my-group")
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
> .setStartingOffsets(OffsetsInitializer.earliest())
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
> .setValueOnlyDeserializer(jsonFormat)
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .build();
> &nbsp; &nbsp; &nbsp; &nbsp; DataStreamSource<Event2&gt; kafkaSource =
> env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");
> &nbsp; &nbsp; &nbsp; &nbsp; kafkaSource.print();
>
>
> &nbsp; &nbsp; &nbsp; &nbsp; // 生成watermark，从数据中提取时间作为事件时间
> &nbsp; &nbsp; &nbsp; &nbsp; SingleOutputStreamOperator<Event2&gt;
> watermarkedStream =
> kafkaSource.assignTimestampsAndWatermarks(WatermarkStrategy.<Event2&gt;forBoundedOutOfOrderness(Duration.ZERO)
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
> .withTimestampAssigner(new SerializableTimestampAssigner<Event2&gt;() {
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
> @Override
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
> public long extractTimestamp(Event2 element, long recordTimestamp) {
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
> &nbsp; &nbsp; SimpleDateFormat simpleDateFormat = new
> SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
> &nbsp; &nbsp; Date date = null;
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
> &nbsp; &nbsp; try {
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
> &nbsp; &nbsp; &nbsp; &nbsp; date =
> simpleDateFormat.parse(element.getTime());
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
> &nbsp; &nbsp; } catch (ParseException e) {
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
> &nbsp; &nbsp; &nbsp; &nbsp; throw new RuntimeException(e);
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
> &nbsp; &nbsp; }
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
> &nbsp; &nbsp; long time = date.getTime();
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
> &nbsp; &nbsp; System.out.println(time);
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
> &nbsp; &nbsp; return time;
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }));
>
>
> &nbsp; &nbsp; &nbsp; &nbsp; // 窗口聚合
> &nbsp; &nbsp; &nbsp; &nbsp; watermarkedStream.map(new MapFunction<Event2,
> Tuple2<String, Long&gt;&gt;() {
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
> @Override
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
> public Tuple2<String, Long&gt; map(Event2 value) throws Exception {
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
> &nbsp; &nbsp; // 将数据转换成二元组，方便计算
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
> &nbsp; &nbsp; return Tuple2.of(value.getUser(), 1L);
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; })
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .keyBy(r -&gt;
> r.f0)
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; // 设置滚动事件时间窗口
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
> .window(TumblingEventTimeWindows.of(Time.seconds(5)))
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .reduce(new
> ReduceFunction<Tuple2<String, Long&gt;&gt;() {
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
> @Override
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
> public Tuple2<String, Long&gt; reduce(Tuple2<String, Long&gt; value1,
> Tuple2<String, Long&gt; value2) throws Exception {
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
> &nbsp; &nbsp; // 定义累加规则，窗口闭合时，向下游发送累加结果
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
> &nbsp; &nbsp; return Tuple2.of(value1.f0, value1.f1 + value2.f1);
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; })
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .print("Aggregated
> stream");
>
>
> &nbsp; &nbsp; &nbsp; &nbsp; env.execute();
> &nbsp; &nbsp; }
> }
>
>
>
>
>
>
> 值得注意的是，若将代码中的 TumblingEventTimeWindows 替换为 TumblingProcessingTimeWindows
> ，即使使用 Kafka 数据源也是可以完成聚合计算并输出结果的。
>
>
>
> 感谢您花时间查看这个问题！
> Lucas