41、Flink 的默认窗口触发器 EventTime 代码示例

1、ContinuousEventTimeTrigger

根据间隔时间周期性触发窗口或者当 Window 的结束时间小于当前的 watermark 时触发窗口计算。

ContinuousEventTimeTrigger.of(Duration.ofSeconds(3))

2、EventTimeTrigger

EventTimeWindows 默认使用，会在 watermark 越过窗口结束时间后直接触发。

3、完整代码示例

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.triggers.ContinuousEventTimeTrigger;
import org.apache.flink.streaming.api.windowing.triggers.ContinuousProcessingTimeTrigger;
import org.apache.flink.streaming.api.windowing.triggers.CountTrigger;
import org.apache.flink.streaming.api.windowing.triggers.EventTimeTrigger;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import java.time.Duration;
public class _10_WindowTriggerDefaultEventTime {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<String> input = env.socketTextStream("localhost", 8888);
// 测试时限制了分区数，生产中需要设置空闲数据源
env.setParallelism(2);
// 事件时间需要设置水位线策略和时间戳
SingleOutputStreamOperator<Tuple2<String, Long>> map = input.map(new MapFunction<String, Tuple2<String, Long>>() {
@Override
public Tuple2<String, Long> map(String input) throws Exception {
String[] fields = input.split(",");
return new Tuple2(fields[0], Long.parseLong(fields[1]));
}
});
SingleOutputStreamOperator<Tuple2<String, Long>> watermarks = map.assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple2<String, Long>>forBoundedOutOfOrderness(Duration.ofSeconds(0))
.withTimestampAssigner(new SerializableTimestampAssigner<Tuple2<String, Long>>() {
@Override
public long extractTimestamp(Tuple2<String, Long> input, long l) {
return input.f1;
}
}));
// ContinuousEventTimeTrigger（根据间隔时间周期性触发窗口或者当 Window 的结束时间小于当前的 watermark 时触发窗口计算）
//a,1718089200000
//b,1718089200000
//c,1718089200000
//
//a,1718089203000
//b,1718089203000
//
//1718089200000-1718089205000
//2> a
//2> a
//1718089200000-1718089205000
//1> b
//1> b
//1718089200000-1718089205000
//1> c
//
//c,1718089203000
//
//a,1718089205000
//b,1718089205000
//
//1718089200000-1718089205000
//2> a
//2> a
//1718089200000-1718089205000
//1> b
//1> b
//1718089200000-1718089205000
//1> c
//1> c
//
//1718089200000-1718089205000
//2> a
//2> a
//1718089200000-1718089205000
//1> b
//1> b
//1718089200000-1718089205000
//1> c
//1> c
//
//c,1718089205000
// EventTimeTrigger（EventTimeWindows 默认使用，会在 watermark 越过窗口结束时间后直接触发）
//a,1718089200000
//b,1718089200000
//c,1718089200000
//
//a,1718089203000
//b,1718089203000
//c,1718089203000
//
//a,1718089205000
//b,1718089205000
//
//1718089200000-1718089205000
//2> a
//2> a
//1718089200000-1718089205000
//1> b
//1> b
//1718089200000-1718089205000
//1> c
//1> c
//
//c,1718089205000
watermarks.keyBy(e -> e.f0)
.window(TumblingEventTimeWindows.of(Duration.ofSeconds(5)))
.trigger(ContinuousEventTimeTrigger.of(Duration.ofSeconds(3)))
//                .trigger(EventTimeTrigger.create())
.apply(new WindowFunction<Tuple2<String, Long>, String, String, TimeWindow>() {
@Override
public void apply(String s, TimeWindow timeWindow, Iterable<Tuple2<String, Long>> iterable, Collector<String> collector) throws Exception {
System.out.println(timeWindow.getStart() + "-" + timeWindow.getEnd());
for (Tuple2<String, Long> tuple : iterable) {
collector.collect(tuple.f0);
}
}
}).print();
env.execute();
}
}