You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flink.apache.org by wei_yuze <we...@qq.com> on 2023/02/07 05:43:59 UTC
Unable to do event time window aggregation with Kafka source
Hello！
I was unable to do event time window aggregation with Kafka source, but had no problem with "fromElement" source. The code is attached as follow. The code has two data sources, named "streamSource" and "kafkaSource" respectively. The program works well with "streamSource", but not with "watermarkedStream".
public class WindowReduceTest2 { public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 使用fromElement数据源
DataStreamSource<Event2> streamSource = env.fromElements(
new Event2("Alice", "./home", "2023-02-04 17:10:11"),
new Event2("Bob", "./cart", "2023-02-04 17:10:12"),
new Event2("Alice", "./home", "2023-02-04 17:10:13"),
new Event2("Alice", "./home", "2023-02-04 17:10:15"),
new Event2("Cary", "./home", "2023-02-04 17:10:16"),
new Event2("Cary", "./home", "2023-02-04 17:10:16")
);
// 使用Kafka数据源
JsonDeserializationSchema<Event2> jsonFormat = new JsonDeserializationSchema<>(Event2.class);
KafkaSource<Event2> source = KafkaSource.<Event2>builder()
.setBootstrapServers(Config.KAFKA_BROKERS)
.setTopics(Config.KAFKA_TOPIC)
.setGroupId("my-group")
.setStartingOffsets(OffsetsInitializer.earliest())
.setValueOnlyDeserializer(jsonFormat)
.build();
DataStreamSource<Event2> kafkaSource = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");
kafkaSource.print();
// 生成watermark，从数据中提取时间作为事件时间
SingleOutputStreamOperator<Event2> watermarkedStream = kafkaSource.assignTimestampsAndWatermarks(WatermarkStrategy.<Event2>forBoundedOutOfOrderness(Duration.ZERO)
.withTimestampAssigner(new SerializableTimestampAssigner<Event2>() {
@Override
public long extractTimestamp(Event2 element, long recordTimestamp) {
SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
Date date = null;
try {
date = simpleDateFormat.parse(element.getTime());
} catch (ParseException e) {
throw new RuntimeException(e);
}
long time = date.getTime();
System.out.println(time);
return time;
}
}));
// 窗口聚合
watermarkedStream.map(new MapFunction<Event2, Tuple2<String, Long>>() {
@Override
public Tuple2<String, Long> map(Event2 value) throws Exception {
// 将数据转换成二元组，方便计算
return Tuple2.of(value.getUser(), 1L);
}
})
.keyBy(r -> r.f0)
// 设置滚动事件时间窗口
.window(TumblingEventTimeWindows.of(Time.seconds(5)))
.reduce(new ReduceFunction<Tuple2<String, Long>>() {
@Override
public Tuple2<String, Long> reduce(Tuple2<String, Long> value1, Tuple2<String, Long> value2) throws Exception {
// 定义累加规则，窗口闭合时，向下游发送累加结果
return Tuple2.of(value1.f0, value1.f1 + value2.f1);
}
})
.print("Aggregated stream");
env.execute();
}
}
Notably, if TumblingEventTimeWindows was replaced with TumblingProcessingTimeWindows, the program works well even with "watermarkedStream"
Thanks for your time!
Lucas
Re: Unable to do event time window aggregation with Kafka source
Posted by yuxia <lu...@alumni.sjtu.edu.cn>.
Hi, Lucas.
What do you mean by saying "unable to do event time window aggregation with watermarkedStream"?
What exception it will throw?
Best regards,
Yuxia
发件人: "wei_yuze" <we...@qq.com>
收件人: "User" <us...@flink.apache.org>
发送时间: 星期二, 2023年 2 月 07日 下午 1:43:59
主题: Unable to do event time window aggregation with Kafka source
Hello！
I was unable to do event time window aggregation with Kafka source, but had no problem with "fromElement" source. The code is attached as follow. The code has two data sources, named "streamSource" and "kafkaSource" respectively. The program works well with "streamSource", but not with "watermarkedStream".
public class WindowReduceTest2 {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 使用fromElement数据源
DataStreamSource<Event2> streamSource = env.fromElements(
new Event2("Alice", "./home", "2023-02-04 17:10:11"),
new Event2("Bob", "./cart", "2023-02-04 17:10:12"),
new Event2("Alice", "./home", "2023-02-04 17:10:13"),
new Event2("Alice", "./home", "2023-02-04 17:10:15"),
new Event2("Cary", "./home", "2023-02-04 17:10:16"),
new Event2("Cary", "./home", "2023-02-04 17:10:16")
);
// 使用Kafka数据源
JsonDeserializationSchema<Event2> jsonFormat = new JsonDeserializationSchema<>(Event2.class);
KafkaSource<Event2> source = KafkaSource.<Event2>builder()
.setBootstrapServers(Config.KAFKA_BROKERS)
.setTopics(Config.KAFKA_TOPIC)
.setGroupId("my-group")
.setStartingOffsets(OffsetsInitializer.earliest())
.setValueOnlyDeserializer(jsonFormat)
.build();
DataStreamSource<Event2> kafkaSource = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");
kafkaSource.print();
// 生成watermark，从数据中提取时间作为事件时间
SingleOutputStreamOperator<Event2> watermarkedStream = kafkaSource.assignTimestampsAndWatermarks(WatermarkStrategy.<Event2>forBoundedOutOfOrderness(Duration.ZERO)
.withTimestampAssigner(new SerializableTimestampAssigner<Event2>() {
@Override
public long extractTimestamp(Event2 element, long recordTimestamp) {
SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
Date date = null;
try {
date = simpleDateFormat.parse(element.getTime());
} catch (ParseException e) {
throw new RuntimeException(e);
}
long time = date.getTime();
System.out.println(time);
return time;
}
}));
// 窗口聚合
watermarkedStream.map(new MapFunction<Event2, Tuple2<String, Long>>() {
@Override
public Tuple2<String, Long> map(Event2 value) throws Exception {
// 将数据转换成二元组，方便计算
return Tuple2.of(value.getUser(), 1L);
}
})
.keyBy(r -> r.f0)
// 设置滚动事件时间窗口
.window(TumblingEventTimeWindows.of(Time.seconds(5)))
.reduce(new ReduceFunction<Tuple2<String, Long>>() {
@Override
public Tuple2<String, Long> reduce(Tuple2<String, Long> value1, Tuple2<String, Long> value2) throws Exception {
// 定义累加规则，窗口闭合时，向下游发送累加结果
return Tuple2.of(value1.f0, value1.f1 + value2.f1);
}
})
.print("Aggregated stream");
env.execute();
}
}
Notably, if TumblingEventTimeWindows was replaced with TumblingProcessingTimeWindows, the program works well even with "watermarkedStream"
Thanks for your time!
Lucas