阅读 15

Flink 的一些概念备忘

Flink:

三种窗口:
1,滚动,就是 5分钟,再5分钟;
2,滑动窗口, 固定一个长度,然后slide,会有很多overlap,比如 5分钟长度, 1分钟滑动。
3,会话窗口,根据多久没有收到数据比如 2分钟,则关闭窗口。直到新的数据,开启新窗口。

Trigger:
窗口触发:
Trigger 提供个三个有用的函数:
1,onElement, onProccessTIme, onEventTime,根据自己需求判断是否触发窗口计算
比如,可以根据count 来算。
也可以根据 时间, 或者watermark 的大小来计算,可以实现非常灵活的计算。

waterMark的机制就是在这里实现。

    @Override
    public TriggerResult onElement(Object element, long timestamp, TimeWindow window, TriggerContext ctx) {
        if (window.maxTimestamp() <= ctx.getCurrentWatermark()) {
            return TriggerResult.FIRE;
        } else {
            ctx.registerEventTimeTimer(window.maxTimestamp());
            return TriggerResult.CONTINUE;
        }
    }
    
    public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) {
    return time == window.maxTimestamp() ?
            TriggerResult.FIRE :
            TriggerResult.CONTINUE;
}

Evictor:
提供一些能力,在窗口触发后,但是在窗口函数计算前后, 做一些定制化的事情,比如对窗口中元素按需进行修剪。flink 默认提供的一些Evictor。
CountEvictor:在窗口维护用户指定数量的元素,如果多于用户指定的数量,从窗口缓冲区的开头丢弃多余的元素。
DeltaEvictor:使用 DeltaFunction 和一个阈值,来计算窗口缓冲区中的最后一个元素与其余每个元素之间的差值,并删除差值大于或等于阈值的元素。
TimeEvictor:以毫秒为单位的时间间隔(interval)作为参数,对于给定的窗口,找到元素中的最大的时间戳max_ts,并删除时间戳小于max_ts - interval的所有元素。

水位线:watermark
水位线为插入到 流记录里面的特殊标识, 一般是用水位线来做窗口触发的标识,
比如9:00到10点的窗口,允许10分钟的延迟, 参看写的文章: https://www.jianshu.com/p/a16dd26f96c6
一般就是 水位线 == 窗口末端就 触发计算。 watermark = max(eventTime- delay)

State 的概念:
每个计算算子,有时候是需要记录状态,比如 source 可能需要记录kafka的offset, 一些聚合算子通常是 增量计算的,也可能在计算你的时候记录 一些状态。
对于checkpoints 等等也需要记录当前的一些状态,以便从错误中恢复。
另外一个用的比较多的,就是broadcast, 我们可以 通过流broadcast 一些配置信息到 stream 中, 然后在processElemnt中获取响应的broadcast内容。
这些broadcast 就是存在 state中,通过 StateDescriptor 就可以获取。

        final MapStateDescriptor<Integer, String> stateDesc = new MapStateDescriptor<>(
                "broadcast-state", Integer.class, String.class
        );

        BroadcastStream<Tuple2<Integer, String>> broadcastStream = ruleStream.broadcast(stateDesc);

        //然后通过
        ctx.getBroadcastState(descriptor)
        

对于flink,有三种存储state 的组件可以选择, inmemory, FileSystem, rocketdb。

对于state来说,通常分为 operator state 和 KeyState
对于operator state, 比如我们可以定义为每个 map 做state的记录,或者为 map operate实现一些checkpoint函数。
checkpoint 函数就可以从我们的 state中进行恢复。

对于keyState,通常可以对KeyedStream 的聚合状态进行 state记录, 然后在state做自己需要的操作,通常数据量会比较大。
在keyState的模型中,还提供很多计算范式,比如TTLState, AggState, ReduceState, 对于 state 的数据是可以按需做相应的操作的。

Partition:

由于并发设置,各个算子之间的并行度 等的设置,需要有某中策略,把 上游对的数据, 分配到下游的机器上,不好的分配策略可能会造成数据倾斜。
其中ForwardPartitioner和GlobalPartitioner两个实现器基本一样,是将记录转发给在本地运行的下游的(归属于subtask)的operation
ShufflePartitioner是随机选择一个channel
RebalancePartitioner实现了一个轮询分区算法
BroadcastPartitioner是将数据发往下游所有节点
RescalPartitioner是通过轮询的方式发往下游

我们完全可以自定义分区器,比如按照某个key的hash进行分区,完全是看自己的业务需求。

作者:NazgulSun

原文链接:https://www.jianshu.com/p/507675adafc1

文章分类
后端
版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌抄袭侵权/违法违规的内容, 请发送邮件至 gxwowoo@163.com 举报,一经查实,本站将立刻删除。
相关推荐