关于flink滚动窗口下数据乱序+倾斜，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。

扩展：

** 在**allowedLateness允许迟到范围内如果有大量数据迟到，因为每来一条迟到数据就触发一次计算，服务器负载会特别高。这就需要重写以下代码(目前博主还没有研究如何修改)：

        if (window.maxTimestamp() <= ctx.getCurrentWatermark()) {
            return TriggerResult.FIRE_AND_PURGE;
        }