目录
前言
滚动窗口（Tumbling Windows）
allowedLateness
场景描述
数据倾斜问题解决
输出结果偏差问题
思考
输出结果偏差解决
扩展
前言
滚动窗口（Tumbling Windows）
滚动窗口的 assigner 分发元素到指定大小的窗口。滚动窗口的大小是固定的，且各自范围之间不重叠。 比如说，如果你指定了滚动窗口的大小为 5 分钟，那么每 5 分钟就会有一个窗口被计算。
allowedLateness
在使用 event-time 窗口时，数据可能会迟到，即 Flink 用来追踪 event-time 进展的 watermark 已经 越过了窗口结束的 timestamp 后，数据才到达。默认情况下，watermark 一旦越过窗口结束的 timestamp，迟到的数据就会被直接丢弃。 但是 Flink 允许指定窗口算子最大的 allowed lateness。 Allowed lateness 定义了一个元素可以在迟到多长时间的情况下不被丢弃，这个参数默认是 0。 在 watermark 超过窗口末端、到达窗口末端加上 allowed lateness 之前的这段时间内到达的元素， 依旧会被加入窗口。取决于窗口的 trigger，一个迟到但没有被丢弃的元素可能会再次触发窗口，比如 EventTimeTrigger。
可以像下面这样指定 allowed lateness
input
.keyBy(<key selector>)
.window(<window assigner>)
.allowedLateness(<time>)
.<windowed transformation>(<window function>);场景描述：
上游数据有一定程度的乱序、并经过keyby之后会有数据倾斜的现象。
数据倾斜问题解决：
1、mapFuncation中的对key进行散列(窗口前)。
Joiner.on("_").join(data[0], new Random().nextInt(3))2、还原key，并进行二次keyby(窗口后)。
aggregate.map(new MapFunction<Tuple3<Long, Integer, String>, Tuple3<Long, Integer, String>>() {
@Override
//in:windowTime,counter,randomKey
//out:windowTime,counter,id
public Tuple3<Long, Integer, String> map(Tuple3<Long, Integer, String> value) throws Exception {
String[] randomKey = value.f2.split("_");
String id = randomKey[0];
return Tuple3.of(value.f0, value.f1, id);
}
}).keyBy(new KeySelector<Tuple3<Long, Integer, String>, Tuple2<Long, String>>() {
@Override
public Tuple2<Long, String> getKey(Tuple3<Long, Integer, String> value) throws Exception {
//tuple2<windowEndTs,id>
return Tuple2.of(value.f0, value.f2);
}
})3、之后使用KeyedProcessFunction进行counter。
输出结果偏差问题
由于使用了allowedLateness、造成最终在KeyedProcessFunction中的汇总结果与预期偏差
思考
下面是eventTime默认触发器源码
@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;
}
}
这是因为使用了allowedLateness之后，窗口内的状态信息不会在窗口关闭时间清除，而是在watemark到达allowedLateness之后再清除。
这个onElement方法是来一个事件会执行一次，其中window.mapTimeStamp就是当前数据的eventTime的最大窗口时间。可以看到当它<=当前水位线，就触发窗口计算，但没有清除窗口内状态信息。因为我们是数据倾斜+数据乱序的场景。后面我们还会有个KeyedProcessFunction进行最终的聚合计算。这就造成了数据重复计算。
TriggerResult有四个属性：
CONTINUE (什么也不做)
FIRE_AND_PURGE (触发计算并清除窗口内状态信息)
FIRE (触发计算)
PURGE (清除窗口内状态信息)
输出结果偏差解决
直接上代码
public class EventTimePurgeTrigger extends Trigger<Object, TimeWindow> {
@Override
public TriggerResult onElement(Object element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
if (window.maxTimestamp() <= ctx.getCurrentWatermark()) {
return TriggerResult.FIRE_AND_PURGE;
} else {
ctx.registerEventTimeTimer(window.maxTimestamp());
return TriggerResult.CONTINUE;
}
}
@Override
public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
return TriggerResult.CONTINUE;
}
@Override
public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
return time == window.maxTimestamp() ? TriggerResult.FIRE_AND_PURGE : TriggerResult.CONTINUE;
}
@Override
public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
ctx.deleteEventTimeTimer(window.maxTimestamp());
}
}在onElement中和onEvenetTime中使用TriggerResult.FIRE_AND_PURGE。文章来源：https://www.toymoban.com/news/detail-832816.html
扩展：
在allowedLateness允许迟到范围内如果有大量数据迟到，因为每来一条迟到数据就触发一次计算，服务器负载会特别高。这就需要重写以下代码(目前博主还没有研究如何修改)：文章来源地址https://www.toymoban.com/news/detail-832816.html
if (window.maxTimestamp() <= ctx.getCurrentWatermark()) {
return TriggerResult.FIRE_AND_PURGE;
} 到了这里，关于关于flink滚动窗口下数据乱序+倾斜，allowedLateness的一个坑的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
