网站的构成元素,如何进行关键词分析,网站在线咨询模块,二维码制作平台前言
Flink 作业的基本骨架结构包含三部分#xff1a;创建执行环境、定义数据处理逻辑、提交并执行Flink作业。
日常大部分 Flink 作业是基于时间窗口计算模型的#xff0c;同样的#xff0c;开发一个Flink时间窗口作业也有一套基本的骨架结构#xff0c;了解这套结构有助…前言
Flink 作业的基本骨架结构包含三部分创建执行环境、定义数据处理逻辑、提交并执行Flink作业。
日常大部分 Flink 作业是基于时间窗口计算模型的同样的开发一个Flink时间窗口作业也有一套基本的骨架结构了解这套结构有助于我们更快地上手时间窗口作业开发。
窗口程序的基本骨架
一个Flink时间窗口作业的代码基本骨架如下所示
stream.keyBy(...) - 仅 keyed 窗口需要.window(...) - 必填项assigner[.trigger(...)] - 可选项trigger (省略则使用默认 trigger)[.evictor(...)] - 可选项evictor (省略则不使用 evictor)[.allowedLateness(...)] - 可选项lateness (省略则为 0)[.sideOutputLateData(...)] - 可选项output tag (省略则不对迟到数据使用 side output).reduce/aggregate/apply() - 必填项function[.getSideOutput(...)] - 可选项output tag时间窗口作业对数据逻辑的处理主要包含以下步骤
对数据流进行分组将DataStream装换为KeyedStream指定窗口分配器 WindowAssigner将数据划分到对应的窗口指定窗口触发器 Trigger决定了窗口何时关闭并计算指定窗口移除器 Evictor它可以在窗口计算前后对窗口内的数据进行移除allowedLateness 允许迟到的数据事件时间语义下即使事件时钟到达窗口关闭时间窗口仍会保留一段时间以等待迟到的数据sideOutputLateData 针对窗口关闭后到达的迟到数据可以将其输出到另外一条数据流对计算结果做修正ProcessFunction 窗口内数据的处理函数
时间窗口作业实战
了解了时间窗口作业的基本骨架以及相关组件的作用接下来就实战一把。
如下示例程序数据源每秒会生成2个一百以内的随机数然后数据经过 keyBy 算子分组这里为了简单数据全部划分为一组KeySelector 统一返回 “all”。
分组后窗口分配器将数据划分到对应的窗口。这里基于处理时间语义统一分配10秒大小的时间窗口时间窗口被Flink封装为 TimeWindow 对象包含两个属性分别是起始时间戳和结束时间戳。
一旦有数据进入窗口Trigger#onElement 就会触发返回值决定了Flink如何处理窗口。显然我们的逻辑是时间到达窗口的结束时间窗口就会触发计算并关闭所以我们会注册一个 ProcessingTime 事件窗口结束时间一到Trigger#onProcessingTime 就会触发窗口就会开始计算。
窗口计算前还需要经过移除器Evictor。它有两个方法分别在窗口计算前和计算后调用在这里你可以根据条件移除窗口内无须计算的数据。示例程序中把小于10的数移除掉了。
最终窗口内的数据会交给 ProcessWindowFunction 处理窗口内的数据被Flink封装成迭代器Iterable通过它可以获得所有窗口内的数据。示例程序 中我们所有元素打印出来并求和。
public class TimeWindowStructure {public static void main(String[] args) throws Exception {StreamExecutionEnvironment environment StreamExecutionEnvironment.getExecutionEnvironment();environment.addSource(new SourceFunctionLong() {Overridepublic void run(SourceContextLong sourceContext) throws Exception {while (true) {Threads.sleep(500);sourceContext.collect(ThreadLocalRandom.current().nextLong(100));}}Overridepublic void cancel() {}}).keyBy(i - all)// 窗口分配器.window(new WindowAssignerLong, TimeWindow() {static final long WINDOW_SIZE 10_000L;Overridepublic CollectionTimeWindow assignWindows(Long event, long timestamp, WindowAssignerContext windowAssignerContext) {// 把数据分配到对应的窗口一条数据甚至可以分配到多个窗口// 这里根据处理时间 分配10秒大小的窗口final long processingTime windowAssignerContext.getCurrentProcessingTime();long start processingTime / WINDOW_SIZE * WINDOW_SIZE;long end start WINDOW_SIZE;return List.of(new TimeWindow(start, end));}Overridepublic TriggerLong, TimeWindow getDefaultTrigger(StreamExecutionEnvironment streamExecutionEnvironment) {// 默认触发器废弃了return null;}Overridepublic TypeSerializerTimeWindow getWindowSerializer(ExecutionConfig executionConfig) {// 窗口序列化器return new TimeWindow.Serializer();}Overridepublic boolean isEventTime() {// 是否基于事件时间语义return false;}})// 窗口触发器.trigger(new TriggerLong, TimeWindow() {private long max_register_processing_time 0L;Overridepublic TriggerResult onElement(Long element, long timestamp, TimeWindow timeWindow, TriggerContext triggerContext) throws Exception {// 每个元素进入窗口都会触发该方法 返回结果决定了窗口是否计算或关闭// 我们是根据处理时间窗口结束时间来判断是否触发的所以注册一个处理时间事件即可if (timeWindow.maxTimestamp() max_register_processing_time) {max_register_processing_time timeWindow.maxTimestamp();triggerContext.registerProcessingTimeTimer(max_register_processing_time);}return TriggerResult.CONTINUE;}Overridepublic TriggerResult onProcessingTime(long l, TimeWindow timeWindow, TriggerContext triggerContext) throws Exception {// 窗口计算并清除数据return TriggerResult.FIRE_AND_PURGE;}Overridepublic TriggerResult onEventTime(long l, TimeWindow timeWindow, TriggerContext triggerContext) throws Exception {return null;}Overridepublic void clear(TimeWindow timeWindow, TriggerContext triggerContext) throws Exception {triggerContext.deleteProcessingTimeTimer(timeWindow.maxTimestamp());}})// 窗口移除器.evictor(new EvictorLong, TimeWindow() {Overridepublic void evictBefore(IterableTimestampedValueLong iterable, int i, TimeWindow timeWindow, EvictorContext evictorContext) {// 窗口计算前触发IteratorTimestampedValueLong iterator iterable.iterator();while (iterator.hasNext()) {TimestampedValueLong next iterator.next();Long value next.getValue();if (value 10) {iterator.remove();System.err.println(Evicted: value);}}}Overridepublic void evictAfter(IterableTimestampedValueLong iterable, int i, TimeWindow timeWindow, EvictorContext evictorContext) {// 窗口计算后触发}})// 因为是基于事件时间语义不存在迟到数据所以无须设置 allowedLateness、sideOutputLateData// 窗口处理函数.process(new ProcessWindowFunctionLong, String, String, TimeWindow() {Overridepublic void process(String group, ProcessWindowFunctionLong, String, String, TimeWindow.Context context, IterableLong iterable, CollectorString collector) throws Exception {TimeWindow window context.window();StringBuilder builder new StringBuilder();builder.append([ window.getStart() - window.maxTimestamp() ] elements:);IteratorLong iterator iterable.iterator();Long sum 0L;while (iterator.hasNext()) {Long value iterator.next();sum value;builder.append(value );}builder.append(, sum: sum);System.err.println(builder.toString());}});environment.execute();}
}运行Flink作业控制台输出
Evicted:3
Evicted:6
Evicted:1
[1722665800000-1722665809999] elements:89 17 16 57 94 47 67 98 , sum:485
Evicted:6
Evicted:4
[1722665810000-1722665819999] elements:86 50 71 95 36 10 55 43 96 36 28 87 89 50 53 35 63 95 , sum:1078
Evicted:4
Evicted:8
Evicted:0
[1722665820000-1722665829999] elements:85 20 42 86 46 20 32 45 91 59 57 64 31 67 78 71 28 , sum:922尾巴
了解Flink时间窗口作业的基本骨架结构理清Flink时间窗口的数据流转过程有助于我们更快上手Flink时间窗口作业的开发。
Flink时间窗口作业包含的核心组件有WindowAssigner、Window、Trigger、Evictor、ProcessWindowFunction。
