0


关于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;
        } 
标签: flink 大数据

本文转载自: https://blog.csdn.net/m0_57219272/article/details/135990813
版权归原作者 维克托把把C 所有, 如有侵权,请联系我们删除。

“关于flink滚动窗口下数据乱序+倾斜,allowedLateness的一个坑”的评论:

还没有评论