Flink窗口是一个很重要的功能，其中要知道窗口是如何触发计算？官方提供哪些触发器？如何自定义触发器？请有耐心看完这篇文章，相信定有所收获！
前提知识储备：窗口知识 事件时间 处理时间 水印。
窗口 Windows
一个无限流可以通过 Windows 拆分为有限大小的 "桶"（集合），我们可以对有限的 "桶" 进行计算。
一个窗口包含如下方法 按照Key分区 选择合适的窗口类型 定义触发器 驱逐器 迟到处理等。其中[...] 表示可选。
stream
.keyBy(...) <- keyed versus non-keyed windows
.window(...) <- required: "assigner"
[.trigger(...)] <- optional: "trigger" (else default trigger)
[.evictor(...)] <- optional: "evictor" (else no evictor)
[.allowedLateness(...)] <- optional: "lateness" (else zero)
[.sideOutputLateData(...)] <- optional: "output tag" (else no side output for late data)
.reduce/aggregate/fold/apply() <- required: "function"
[.getSideOutput(...)] <- optional: "output tag"
还有一种是全局窗口，所有数据都放入一个窗口中！
触发器 Trigger
触发器是用来确认窗口什么时候准备完毕，可以用来处理函数处理窗口数据。
简单使用：
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<Integer> source = env.fromElements(1, 3, 5, 9, 20, 100, 100, 200);
source.keyBy(v-> "")
.window(TumblingProcessingTimeWindows.of(Time.of(60, TimeUnit.SECONDS)))
.trigger(ProcessingTimeTrigger.create()) //触发器
.process(new ProcessWindowFunction<Integer, Integer, String, TimeWindow>() {
@Override
public void process(String s, Context context, Iterable<Integer> elements, Collector<Integer> out) throws Exception {
//处理逻辑 ...
}
});
上面定义一个基于处理时间(ProcessingTime)的翻滚窗口(TumblingProcessingTimeWindows)， 声明了基于处理时间触发器(ProcessingTimeTrigger)。
trigger不声明会使用默认触发器 可以查看如下逻辑；
调用路径
org.apache.flink.streaming.api.datastream.KeyedStream#window org.apache.flink.streaming.api.windowing.assigners.WindowAssigner#getDefaultTrigger org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows#getDefaultTrigger org.apache.flink.streaming.api.datastream.WindowedStream#WindowedStream
TumblingProcessingTimeWindows 只是 WindowAssigner 子类；所以不同的窗口类型返回的默认触发器不一样；
触发器事件：
触发器需要结合返回结果来进行窗口触发，所有触发器会返回一个触发结果类型。
Flink提供一个TriggerResult类：
public enum TriggerResult {
/** 不做任何操作 */
CONTINUE(false, false),
/** 触发窗口计算，且清除窗口数据 */
FIRE_AND_PURGE(true, true),
/**
* 只触发窗口计算 不会清理窗口数据
*/
FIRE(true, false),
/**
* 只清理窗口数据 不触发窗口计算
*/
PURGE(false, true);
}
CONTINUE 不做任何操作 FIRE 触发窗口计算 PURGE 清理窗口数据 FIRE_AND_PURGE 触发窗口计算，然后清理窗口数据
触发动作：
Flink中提供了Trigger这个接口，里面定义了核心的方法。
@PublicEvolving
public abstract class Trigger<T, W extends Window> implements Serializable {
public abstract TriggerResult onElement(T element, long timestamp, W window, TriggerContext ctx) throws Exception;
public abstract TriggerResult onProcessingTime(long time, W window, TriggerContext ctx) throws Exception;
public abstract TriggerResult onEventTime(long time, W window, TriggerContext ctx) throws Exception;
}
onElement
添加到窗口的每个元素，调用该方法。
onProcessingTime
注册的处理时间计时器触发时调用该方法。
onEventTime
注册的事件时间计时器触发时调用该方法。
7种触发器
所有触发器都实现了 org.apache.flink.streaming.api.windowing.triggers.Trigger
官方提供了7种触发器：按照类型划分（时间，数量） 按照时间划分（事件时间，处理时间）
PurgingTrigger
可以将任何触发器都变成计算后清除窗口元素，这个触发器可以包装其他触发器。
@PublicEvolving
public class PurgingTrigger<T, W extends Window> extends Trigger<T, W> {
private static final long serialVersionUID = 1L;
private Trigger<T, W> nestedTrigger;
private PurgingTrigger(Trigger<T, W> nestedTrigger) {
this.nestedTrigger = nestedTrigger;
}
@Override
public TriggerResult onElement(T element, long timestamp, W window, TriggerContext ctx) throws Exception {
TriggerResult triggerResult = nestedTrigger.onElement(element, timestamp, window, ctx);
return triggerResult.isFire() ? TriggerResult.FIRE_AND_PURGE : triggerResult;
}
@Override
public TriggerResult onEventTime(long time, W window, TriggerContext ctx) throws Exception {
TriggerResult triggerResult = nestedTrigger.onEventTime(time, window, ctx);
return triggerResult.isFire() ? TriggerResult.FIRE_AND_PURGE : triggerResult;
}
@Override
public TriggerResult onProcessingTime(long time, W window, TriggerContext ctx) throws Exception {
TriggerResult triggerResult = nestedTrigger.onProcessingTime(time, window, ctx);
return triggerResult.isFire() ? TriggerResult.FIRE_AND_PURGE : triggerResult;
}
可以看到内部维护了一个nestedTrigger触发器类 如果nestedTrigger返回是 Fire 就返回 TriggerResult.FIRE_AND_PURGE
实际代码使用：
source.keyBy(v-> "")
.window(TumblingProcessingTimeWindows.of(Time.of(60, TimeUnit.SECONDS)))
.trigger(PurgingTrigger.of(ProcessingTimeTrigger.create())) //触发器
.process(new ProcessWindowFunction<Integer, Integer, String, TimeWindow>() {
@Override
public void process(String s, Context context, Iterable<Integer> elements, Collector<Integer> out) throws Exception {
//处理逻辑
}
});
EventTimeTrigger
事件时间触发器
@PublicEvolving
public class EventTimeTrigger extends Trigger<Object, TimeWindow> {
private static final long serialVersionUID = 1L;
private EventTimeTrigger() {}
@Override
public TriggerResult onElement(Object element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
if (window.maxTimestamp() <= ctx.getCurrentWatermark()) {
// if the watermark is already past the window fire immediately
return TriggerResult.FIRE;
} else {
ctx.registerEventTimeTimer(window.maxTimestamp());
return TriggerResult.CONTINUE;
}
}
@Override
public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) {
return time == window.maxTimestamp() ?
TriggerResult.FIRE :
TriggerResult.CONTINUE;
}
@Override
public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
return TriggerResult.CONTINUE;
}
}
因为这是一个事件时间触发器，所以在onProcessingTime方法中不做任何处理。我们只关注onElement和onEventTime这两个方法。
onElement解析：首先大家都知道 如果是翻滚窗口 假如窗口是一分钟的话 窗口的【开始时间】-【结束时间】如下：开始窗口： 【2001-1-1 00:00:00】-【2001-1-1 00:00:59】
下个窗口： 【2001-1-1 00:01:00】-【2001-1-1 00:01:59】
以此类推！实际中存储的都是时间戳，这里用我们常见的格式方便大家理解。
当一个element元素进入该方法， 会通过 window.maxTimestamp() 获取窗口的结束时间 如果当前水印大于等于窗口的结束时间， 那么代表这个窗口可以触发了。就返回 FIRE比如【2001-1-1 00:00:59】 <= 水印 就触发计算。
如果当前水印还没推进到窗口的最大时间，这个时候就会通过事件时间注册器注册一个窗口最大结束时间定时器。如：会注册一个定时器【2001-1-1 00:00:59】在水印推进到 【2001-1-1 00:00:59】时候回调 onEventTime()
onEventTime解析：当前方法就比较简单了， 在onElement注册的定时器，会在水印大于等于定时器的时间时，触发执行onEventTime方法， 这个方法有三个参数，其中第一个参数 代表之前注册的定时器时间， 比如上文中注册了 【2001-1-1 00:00:59】 那么回调的时候第一个参数就代表【2001-1-1 00:00:59】
如果回调的时间等于窗口结束时间，说明可以触发窗口执行了；否则就不需要触发执行；
个人疑问：为什么触发器是里面的触发时通过等于判断（time == window.maxTimestamp()） ？（time >= window.maxTimestamp()） 是不是可以？
ContinuousEventTimeTrigger
基于水印，根据给定的时间连续触发。
背景是这样子的， 实际场景中可能有很多任务时间窗口高达一天或者一小时，对于业务分析来说它希望尽快能看到当前的结果。所以为了减少延迟，希望窗口提前触发，这个时候就可以考虑使用该窗口。
@PublicEvolving
public class ContinuousEventTimeTrigger<W extends Window> extends Trigger<Object, W> {
private static final long serialVersionUID = 1L;
//触发间隔
private final long interval;
private final ReducingStateDescriptor<Long> stateDesc =
new ReducingStateDescriptor<>("fire-time", new Min(), LongSerializer.INSTANCE);
private ContinuousEventTimeTrigger(long interval) {
this.interval = interval;
}
@Override
public TriggerResult onElement(Object element, long timestamp, W window, TriggerContext ctx) throws Exception {
if (window.maxTimestamp() <= ctx.getCurrentWatermark()) {
// 如果水印大于窗口结束时间 触发结算
return TriggerResult.FIRE;
} else {
//使用窗口结束时间 注册一个事件时间定时器
ctx.registerEventTimeTimer(window.maxTimestamp());
}
// 维护一个单值状态
ReducingState<Long> fireTimestamp = ctx.getPartitionedState(stateDesc);
if (fireTimestamp.get() == null) {
//计算下个触发窗口计算的时间
long start = timestamp - (timestamp % interval);
long nextFireTimestamp = start + interval;
//使用下个触发窗口计算时间注册一个事件时间定时器
ctx.registerEventTimeTimer(nextFireTimestamp);
//保存进状态
fireTimestamp.add(nextFireTimestamp);
}
return TriggerResult.CONTINUE;
}
@Override
public TriggerResult onEventTime(long time, W window, TriggerContext ctx) throws Exception {
//如果触发时间等于窗口结束时间 那么就触发窗口计算
if (time == window.maxTimestamp()){
return TriggerResult.FIRE;
}
ReducingState<Long> fireTimestampState = ctx.getPartitionedState(stateDesc);
//获取触发时间戳
Long fireTimestamp = fireTimestampState.get();
//如果不为空且触发时间等于定时器时间 说明这次触发的是 间隔定时器
if (fireTimestamp != null && fireTimestamp == time) {
//将原来的状态清空 再之前触发时间加上间隔时间重新注册一个定时器，解散此次结果
fireTimestampState.clear();
fireTimestampState.add(time + interval);
ctx.registerEventTimeTimer(time + interval);
return TriggerResult.FIRE;
}
return TriggerResult.CONTINUE;
}
}
ContinuousEventTimeTrigger 比 EventTimeTrigger 多一个参数和状态描述符。interval：间隔时间 stateDesc：状态描述符 用来获取状态
那么间隔时间有什么作用？我们先来看一张图。
通过此图可知 我们有一个窗口，范围是 0-59秒，而其中窗口内有两个子窗口 这个子窗口根据interval计算，
此图中的间隔时间为20秒 ， 这样就可以0-20触发一次 0-40触发一次 0-59触发一次；这样就可以触发多次，
减少延迟时间！
不过注意：窗口内多次计算，下次计算结果包含上一次的值的，所以下游接收根据业务实际情况处理，比如通过幂等写入更新；
不过使用PurgingTrigger触发器包装ContinuousEventTimeTrigger那么下次计算是不包含上次的计算值的；如下图
图：使用了PurgingTrigger的效果
我们先看onElement方法源码：前面逻辑也是 使用水印和窗口结束时间判断 大于等于触发计算，否则注册一个窗口结束时间事件时间定时器！
它比多了几个步骤：
//计算下个触发窗口计算的时间
long start = timestamp - (timestamp % interval);
long nextFireTimestamp = start + interval;
通过 interval 计算下一个时间触发时间；逻辑如下 一个窗口 0-50秒，interval=20秒， 那么它会使用当前元素的事件时间（EventTime） 即代码中的timestamp去求整数， 举个例子 现在有元素 6，27，49 这些数字就代表事件时间； **timestamp=6,27,49 **
首先 元素6进来窗口 它的 timestamp=6
start = 6-（6%20） = 0;
nextFireTimestamp = start + interval = 0 + 20 = 20
然后 元素27进来窗口 它的 timestamp=27 start = 27 -（27%20） = 20; nextFireTimestamp = start + interval = 20 + 20 = 40
然后 元素 49 进来窗口 它的 timestamp=49 start = 49 -（49 %20） = 40; nextFireTimestamp = start + interval = 40+ 20 = 60
所以分别会注册 20秒 40秒 60秒的定时器！
**抛出一个问题给大家思考下：** 如果在一个较短的时间内 注册20秒-40秒的定时器？这个时候怎么玩？ 明明状态只存储一个值！
private final ReducingStateDescriptor<Long> stateDesc =
new ReducingStateDescriptor<>("fire-time", new Min(), LongSerializer.INSTANCE);
重点关注 new Min()
private static class Min implements ReduceFunction<Long> {
private static final long serialVersionUID = 1L;
@Override
public Long reduce(Long value1, Long value2) throws Exception {
return Math.min(value1, value2);
}
}
可以看到 20秒和40秒同时进去，它会取较小的（20秒）作为下一次触发时间。
那我们再来看看定时onEventTime代码如何多次触发：
//如果触发时间等于窗口结束时间 那么就触发窗口计算
if (time == window.maxTimestamp()){
return TriggerResult.FIRE;
}
ReducingState<Long> fireTimestampState = ctx.getPartitionedState(stateDesc);
//获取触发时间戳
Long fireTimestamp = fireTimestampState.get();
//如果不为空且触发时间等于定时器时间 说明这次触发的是 间隔定时器
if (fireTimestamp != null && fireTimestamp == time) {
//将原来的状态清空 再之前触发时间加上间隔时间重新注册一个定时器，解散此次结果
fireTimestampState.clear();
fireTimestampState.add(time + interval);
ctx.registerEventTimeTimer(time + interval);
return TriggerResult.FIRE;
}
return TriggerResult.CONTINUE;
可以看出 如果当前注册的是窗口结束时间（相等）就触发窗口计算；否则去判断当前定时器是不是注册的多次计算定时器，如果是就清空状态；在之前的注册时间加上间隔时间更新状态且重新注册新的多次计算定时器；然后触发窗口内的间隔计算。
举个例子：背景还是上文，之前注册20秒间隔计算，那么就会触发计算，并且20+间隔时间（20秒）= 40秒 然后使用40秒重新注册定时器。
代码栗子：
public class ContinuousEventTimeTriggerDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<Tuple2<String, Long>> source = env.addSource(new SourceFunction<Tuple2<String, Long>>() {
@Override
public void run(SourceContext<Tuple2<String, Long>> ctx) throws Exception {
//手动输入元素 格式李白,1000
Scanner scanner = new Scanner(System.in);
while (true) {
if (scanner.hasNext()) {
String next = scanner.next();
String[] split = next.split(",");
System.out.println("输入了：" + split[0] + " : " + split[1]);
ctx.collect(Tuple2.of(split[0], Long.valueOf(split[1])));
}
}
}
});
//抽取 水印
SingleOutputStreamOperator<Tuple2<String, Long>> streamOperator = source.assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple2<String, Long>>forMonotonousTimestamps().withTimestampAssigner((v, t) -> v.f1));
//窗口统计
SingleOutputStreamOperator<Integer> process = streamOperator.keyBy(v -> v.f0)
.window(TumblingEventTimeWindows.of(Time.of(60, TimeUnit.SECONDS)))
.trigger(ContinuousEventTimeTrigger.of(Time.of(20, TimeUnit.SECONDS)))
.process(new ProcessWindowFunction<Tuple2<String, Long>, Integer, String, TimeWindow>() {
@Override
public void process(String s, Context context, Iterable<Tuple2<String, Long>> elements, Collector<Integer> out) throws Exception {
Integer counter = 0;
for (Iterator<Tuple2<String, Long>> iterator = elements.iterator(); iterator.hasNext(); ) {
Tuple2<String, Long> next = iterator.next();
counter++;
}
out.collect(counter);
System.out.println("STR 触发提醒：" + context.window().getStart() + " - " + context.window().getEnd() + " key: " + s + " count: " + counter);
}
});
process.print();
env.execute("--------------");
}
}
运行结果：
李白,1000
输入了：李白 : 1000
李白,10000
输入了：李白 : 10000
李白,20000
输入了：李白 : 20000
李白,25000
输入了：李白 : 25000
4> 4
STR 触发提醒：0 - 60000 key: 李白 count: 4
李白,45000
输入了：李白 : 45000
4> 5
STR 触发提醒：0 - 60000 key: 李白 count: 5
李白,65000
输入了：李白 : 65000
4> 5
STR 触发提醒：0 - 60000 key: 李白 count: 5
这个栗子就是60秒窗口， 窗口内每20秒触发一次， 通过输出结果可以看到
// 25秒触发了20秒的定时器 输出一次结果
李白,25000
输入了：李白 : 25000
4> 4
STR 触发提醒：0 - 60000 key: 李白 count: 4
// 45秒触发了40秒的定时器 输出一次结果
李白,45000
输入了：李白 : 45000
4> 5
STR 触发提醒：0 - 60000 key: 李白 count: 5
//最后 一个元素大于60秒 窗口结束
李白,65000
输入了：李白 : 65000
4> 5
STR 触发提醒：0 - 60000 key: 李白 count: 5
DeltaTrigger
基于阈值的触发器， 这个触发器可以根据上一个元素和当前元素的阈值（自定义逻辑）触发计算！
PublicEvolving
public class DeltaTrigger<T, W extends Window> extends Trigger<T, W> {
private static final long serialVersionUID = 1L;
//自定义的阈值方法
private final DeltaFunction<T> deltaFunction;
//阈值 double
private final double threshold;
private final ValueStateDescriptor<T> stateDesc;
private DeltaTrigger(double threshold, DeltaFunction<T> deltaFunction, TypeSerializer<T> stateSerializer) {
this.deltaFunction = deltaFunction;
this.threshold = threshold;
stateDesc = new ValueStateDescriptor<>("last-element", stateSerializer);
}
}
重点关注：threshold
它是一个double类型数值！
DeltaFunction
它是一个接口，需要用户自定义逻辑！
@PublicEvolving
public interface DeltaFunction<DATA> extends Serializable {
/**
* Calculates the delta between two given data points.
*
* @param oldDataPoint 旧的值
* @param newDataPoint 新的值
* @return 两个值之间延迟
*/
double getDelta(DATA oldDataPoint, DATA newDataPoint);
}
因为这个触发器是基于元素之间的比较， 所以onEventTime和onProcessingTime不做任何处理！
我们看看 onElement 源码：
@Override
public TriggerResult onElement(T element, long timestamp, W window, TriggerContext ctx) throws Exception {
// lastElementState 存储了窗口内最早元素
ValueState<T> lastElementState = ctx.getPartitionedState(stateDesc);
if (lastElementState.value() == null) {
//初始化状态值
lastElementState.update(element);
return TriggerResult.CONTINUE;
}
//如果通过延迟函数的getDelta方法，计算得到结果大于指定的阈值 就触发计算 且将当前元素更新状态
if (deltaFunction.getDelta(lastElementState.value(), element) > this.threshold) {
lastElementState.update(element);
return TriggerResult.FIRE;
}
return TriggerResult.CONTINUE;
}
逻辑比较简单，就是通过状态存储了最早的元素， 然后通过 DeltaFunction#getDelta 获取计算的延迟值；再根据阈值比较， 大于阈值就触发计算且更新当前元素为最新状态。否则不做任何触发；
举个栗子：
public class DeltaWindowsDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<Integer> source = env.addSource(new SourceFunction<Integer>() {
@Override
public void run(SourceContext<Integer> ctx) throws Exception {
//手动输入元素 格式李白,1000
Scanner scanner = new Scanner(System.in);
while (true) {
if (scanner.hasNext()) {
String next = scanner.next();
System.out.println("输入了：" + next);
ctx.collect(Integer.valueOf(next));
}
}
}
@Override
public void cancel() {
}
});
//抽取 水印
SingleOutputStreamOperator<Integer> streamOperator = source.assignTimestampsAndWatermarks(WatermarkStrategy.<Integer>forMonotonousTimestamps().withTimestampAssigner((v, t) -> v));
//窗口统计
SingleOutputStreamOperator<Integer> process = streamOperator.keyBy(v -> "")
.window(TumblingEventTimeWindows.of(Time.of(60, TimeUnit.SECONDS)))
.trigger(DeltaTrigger.of(20000, new DeltaFunction<Integer>() {
@Override
public double getDelta(Integer oldDataPoint, Integer newDataPoint) {
return newDataPoint - oldDataPoint;
}
}, new IntSerializer()))
.process(new ProcessWindowFunction<Integer, Integer, String, TimeWindow>() {
@Override
public void process(String s, Context context, Iterable<Integer> elements, Collector<Integer> out) throws Exception {
Integer total = 0;
int count = 0;
for (Iterator<Integer> iterator = elements.iterator(); iterator.hasNext(); ) {
Integer next = iterator.next();
total += next;
count++;
}
System.out.println("STR 触发提醒：" + context.window().getStart() + " - " + context.window().getEnd() + "总数：" + total + " 总条数：" + count);
}
});
process.print();
env.execute("--------------");
}
}
打印结果：
10000
输入了：10000
20000
输入了：20000
30000
输入了：30000
40000
输入了：40000
STR 触发提醒：0 - 60000总数：100000 总条数：4
50000
输入了：50000
70000
输入了：70000
71000
输入了：71000
72000
输入了：72000
90000
输入了：90000
91000
输入了：91000
STR 触发提醒：60000 - 120000总数：394000 总条数：5
92000
输入了：92000
110000
输入了：110000
113000
输入了：113000
STR 触发提醒：60000 - 120000总数：709000 总条数：8
样例中按照一分钟生成一个窗口，可以看到一个窗口内我分别输出入了 10000，20000，30000， 40000
也就是10s, 20s, 30s, 40s 可以发现最早的元素是 10000 输入到 40000 时候 满足 （40000 - 10000 ）> 20000阈值，触发第一次计算； 输出了 STR 触发提醒：0 - 60000总数：100000 总条数：4
很大的疑问：这里会发现 50000 没有触发计算了！ 它属于第一个窗口的数据， 但是想要触发50000输出，必须满足输入的元素减去50000大于20000 这个数字是大于70000的！然后我输出大于71000这个元素，按理来说50000要参加触发了，但是它是属于第二窗口的。根本不会进入到第一个窗口里面计算，这个时候就会发现第一个窗口就永远不会再输出了！
所以个人认为这个触发器只有在全局窗口时候有用；
streamOperator.windowAll(GlobalWindows.create())
.trigger(DeltaTrigger.of(20000, new DeltaFunction<Integer>() {
@Override
public double getDelta(Integer oldDataPoint, Integer newDataPoint) {
return newDataPoint - oldDataPoint;
}
}, new IntSerializer()))
.process(new ProcessAllWindowFunction<Integer, Object, GlobalWindow>() {
@Override
public void process(Context context, Iterable<Integer> elements, Collector<Object> out) throws Exception {
Integer total = 0;
int count = 0;
for (Iterator<Integer> iterator = elements.iterator(); iterator.hasNext(); ) {
Integer next = iterator.next();
total += next;
count++;
}
System.out.println("STR 触发提醒：" + "总数：" + total + " 总条数：" + count);
}
});
暂时没有找到DeltaTrigger其它的使用场景！大家有想法可以交流。
CountTrigger
当窗口元素达到给定的计数就触发计算；
@PublicEvolving
public class CountTrigger<W extends Window> extends Trigger<Object, W> {
private static final long serialVersionUID = 1L;
//最大元素个数
private final long maxCount;
// 状态描述符
private final ReducingStateDescriptor<Long> stateDesc =
new ReducingStateDescriptor<>("count", new Sum(), LongSerializer.INSTANCE);
private CountTrigger(long maxCount) {
this.maxCount = maxCount;
}
}
这里面有一个最大元素个数 maxCount
也相当于一个阈值！
其中里面也有一个单值的聚合状态 ReducingState
通过 ReducingStateDescriptor
可知将新增的状态值通过new Sum()
处理！
private static class Sum implements ReduceFunction<Long> {
private static final long serialVersionUID = 1L;
@Override
public Long reduce(Long value1, Long value2) throws Exception {
return value1 + value2;
}
}
通过代码可知这里做一个新值和旧值相加并返回出去！
我们来瞧一瞧onElement源码：
@Override
public TriggerResult onElement(Object element, long timestamp, W window, TriggerContext ctx) throws Exception {
ReducingState<Long> count = ctx.getPartitionedState(stateDesc);
count.add(1L);
if (count.get() >= maxCount) {
count.clear();
return TriggerResult.FIRE;
}
return TriggerResult.CONTINUE;
}
代码尤其简单
首先获取状态且状态count.add(1L)
这个就相当于调用了Sum.reduce(1,1)
当历史没有值时，不会调用。然后从状态取值跟 maxCount
判断，如果大于等于就是触发计算，并清楚状态； 否则就不做任何处理！
其实通过这里通过上面 DeltaTrigger 结果可知，在非全局窗口中使用，会造成部分窗口元素不触发计算！所以这里应该也是用来和全局窗口做判断的！
举个栗子：
public class CountTriggerDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<Integer> source = env.addSource(new SourceFunction<Integer>() {
@Override
public void run(SourceContext<Integer> ctx) throws Exception {
//手动输入元素 格式李白,1000
Scanner scanner = new Scanner(System.in);
while (true) {
if (scanner.hasNext()) {
String next = scanner.next();
System.out.println("输入了：" + next);
ctx.collect(Integer.valueOf(next));
}
}
}
@Override
public void cancel() {
}
});
//抽取 水印
SingleOutputStreamOperator<Integer> streamOperator = source.assignTimestampsAndWatermarks(WatermarkStrategy.<Integer>forMonotonousTimestamps().withTimestampAssigner((v, t) -> v));
streamOperator.windowAll(GlobalWindows.create())
.trigger(CountTrigger.of(3))
.process(new ProcessAllWindowFunction<Integer, Object, GlobalWindow>() {
@Override
public void process(Context context, Iterable<Integer> elements, Collector<Object> out) throws Exception {
Integer total = 0;
int count = 0;
for (Iterator<Integer> iterator = elements.iterator(); iterator.hasNext(); ) {
Integer next = iterator.next();
total += next;
count++;
}
System.out.println("STR 触发提醒：" + "总数：" + total + " 总条数：" + count);
}
});
env.execute("222");
}
}
打印结果：
1
输入了：1
2
输入了：2
3
输入了：3
STR 触发提醒：总数：6 总条数：3
4
输入了：4
5
输入了：5
6
输入了：6
STR 触发提醒：总数：21 总条数：6
7
输入了：7
8
输入了：8
9
输入了：9
STR 触发提醒：总数：45 总条数：9
可以看到每写入三条数据！都会触发一次计算！
ProcessingTimeTrigger 和 ContinuousProcessingTimeTrigger
基于处理时间的触发器，这两个触发器很简单，我就不做过多的解读了，大家可以自己看看源码！ContinuousProcessingTimeTrigger 可以参考 ContinuousEventTimeTrigger 就是定时器换成了 onProcessingTime 逻辑基本一样。
自定义触发器
这里带大家实现一个 基于事件时间 每进来一个元素触发一次计算触发器！
我们首先定义触发器类 实现触发器的必要方法，可以看到onElement每进来一个元素，就触发一次！然后定时器触发一次。
public class EveryElementEventTimeTrigger<W extends Window> extends Trigger<Object, W> {
@Override
public TriggerResult onElement(Object element, long timestamp, W window, TriggerContext ctx) throws Exception {
ctx.registerProcessingTimeTimer(window.maxTimestamp());
return TriggerResult.FIRE;
}
@Override
public TriggerResult onProcessingTime(long time, W window, TriggerContext ctx) throws Exception {
return TriggerResult.CONTINUE;
}
@Override
public TriggerResult onEventTime(long time, W window, TriggerContext ctx) throws Exception {
return TriggerResult.FIRE;
}
@Override
public void clear(W window, TriggerContext ctx) throws Exception {
//如果维护了状态， 也可以在这里做状态清理！
ctx.deleteEventTimeTimer(window.maxTimestamp());
}
public static EveryElementEventTimeTrigger create() {
return new EveryElementEventTimeTrigger();
}
}
使用：
public class EveryElementDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<Tuple2<String, Long>> source = env.addSource(new SourceFunction<Tuple2<String, Long>>() {
@Override
public void run(SourceContext<Tuple2<String, Long>> ctx) throws Exception {
//手动输入 name,1000
Scanner scanner = new Scanner(System.in);
while (true) {
if (scanner.hasNext()) {
String next = scanner.next();
String[] split = next.split(",");
System.out.println("输入了：" + split[0] + " : " + split[1]);
ctx.collect(Tuple2.of(split[0], Long.valueOf(split[1])));
}
}
}
@Override
public void cancel() {
}
});
//水印抽取
SingleOutputStreamOperator<Tuple2<String, Long>> streamOperator = source.assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple2<String, Long>>forMonotonousTimestamps().withTimestampAssigner((v, t) -> v.f1));
SingleOutputStreamOperator<Integer> process = streamOperator.keyBy(v -> "")
.window(TumblingEventTimeWindows.of(Time.of(60, TimeUnit.SECONDS)))
.trigger(EveryElementEventTimeTrigger.create())
.process(new ProcessWindowFunction<Tuple2<String, Long>, Integer, String, TimeWindow>() {
@Override
public void process(String s, Context context, Iterable<Tuple2<String, Long>> elements, Collector<Integer> out) throws Exception {
Integer counter = 0;
for (Iterator<Tuple2<String, Long>> iterator = elements.iterator(); iterator.hasNext(); ) {
Tuple2<String, Long> next = iterator.next();
counter++;
}
out.collect(counter);
System.out.println("STR 触发提醒：" + context.window().getStart() + " - " + context.window().getEnd() + " 总数：" + counter);
}
});
process.print();
env.execute("----Job----");
}
}
输出结果：
zk,1000
输入了：zk : 1000
5> 1
STR 触发提醒：0 - 60000 总数：1
zk,10000
输入了：zk : 10000
5> 2
STR 触发提醒：0 - 60000 总数：2
zk,30000
输入了：zk : 30000
5> 3
STR 触发提醒：0 - 60000 总数：3
zk,60000
输入了：zk : 60000
5> 1
STR 触发提醒：60000 - 120000 总数：1
5> 3
STR 触发提醒：0 - 60000 总数：3
窗口触发核心源码
Flink又是在什么时候调用触发器的呢？
返回的TriggerResult有什么作用？都可以这里找到答案，我省略部分代码，欢迎品鉴！
@Internal
public class WindowOperator<K, IN, ACC, OUT, W extends Window>
extends AbstractUdfStreamOperator<OUT, InternalWindowFunction<ACC, OUT, K, W>>
implements OneInputStreamOperator<IN, OUT>, Triggerable<K, W> {
@Override
public void processElement(StreamRecord<IN> element) throws Exception {
final Collection<W> elementWindows = windowAssigner.assignWindows(
element.getValue(), element.getTimestamp(), windowAssignerContext);
...
for (W window: elementWindows) {
// drop if the window is already late
if (isWindowLate(window)) {
continue;
}
isSkippedElement = false;
windowState.setCurrentNamespace(window);
windowState.add(element.getValue());
triggerContext.key = key;
triggerContext.window = window;
//调用触发器的onElement方法
TriggerResult triggerResult = triggerContext.onElement(element);
//如果触发器结果是计算
if (triggerResult.isFire()) {
ACC contents = windowState.get();
if (contents == null) {
continue;
}
//调用用户的处理方法将窗口发送出去计算
emitWindowContents(window, contents);
}
//如果触发器结果是清除 就清除窗口状态
if (triggerResult.isPurge()) {
windowState.clear();
}
registerCleanupTimer(window);
}
...
}
@Override
public void onEventTime(InternalTimer<K, W> timer) throws Exception {
...
//调用触发器的onEventTime方法
TriggerResult triggerResult = triggerContext.onEventTime(timer.getTimestamp());
//如果触发器结果是计算
if (triggerResult.isFire()) {
ACC contents = windowState.get();
if (contents != null) {
//调用用户的处理法将窗口发送出去计算
emitWindowContents(triggerContext.window, contents);
}
}
//如果触发器结果是清除 就清除窗口状态
if (triggerResult.isPurge()) {
windowState.clear();
}
if (windowAssigner.isEventTime() && isCleanupTime(triggerContext.window, timer.getTimestamp())) {
clearAllState(triggerContext.window, windowState, mergingWindows);
}
...
}
@Override
public void onProcessingTime(InternalTimer<K, W> timer) throws Exception {
...
//调用触发器的onProcessingTime方法
TriggerResult triggerResult = triggerContext.onProcessingTime(timer.getTimestamp());
//如果触发器结果是计算
if (triggerResult.isFire()) {
ACC contents = windowState.get();
if (contents != null) {
//调用用户的处理法将窗口发送出去计算
emitWindowContents(triggerContext.window, contents);
}
}
//如果触发器结果是清除 就清除窗口状态
if (triggerResult.isPurge()) {
windowState.clear();
}
if (!windowAssigner.isEventTime() && isCleanupTime(triggerContext.window, timer.getTimestamp())) {
clearAllState(triggerContext.window, windowState, mergingWindows);
}
...
}
}
总结
我们开始回顾了窗口使用方式，和窗口的几种类型；又带大家认识了窗口中触发器的作用，和触发器返回的枚举值作用；我们通过源码加实践的方式学习了，官方提供的7种触发器，而且知道了如何搭配使用，避免大家踩坑！最后通过一个自定义触发器教大家如何实现自己的触发器。
文章转载自OfNull，如果涉嫌侵权，请发送邮件至：contact@modb.pro进行举报，并提供相关证据，一经查实，墨天轮将立刻删除相关内容。
