Flink 1.14.* Flink窗口创建和窗口计算源码

2024-08-30 05:12
文章标签 源码 计算 创建 窗口 flink 1.14

本文主要是介绍Flink 1.14.* Flink窗口创建和窗口计算源码，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

解析Flink如何创建的窗口，和以聚合函数为例，窗口如何计算聚合函数

  • 一、构建不同窗口的build类
    • 1、全局窗口
    • 2、创建按键分流后的窗口
  • 二、在使用窗口处理数据流时，不同窗口创建的都是窗口算子WindowOperator
    • 1、聚合函数实现
    • 2、创建全局窗口(入参传的是NullByteKeySelector)
    • 3、创建按键分流后的窗口(入参传的是KeyedStream的KeySelector)
    • 3、WindowOperator

一、构建不同窗口的build类

这个是示例，

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<Tuple2<String, Integer>> input = env.fromElements(Tuple2.of("key1", 1),Tuple2.of("key1", 3),Tuple2.of("key2", 2),Tuple2.of("key2", 4));

1、全局窗口

下面是创建全局窗口的代码

AllWindowedStream<Tuple2<String, Integer>, TimeWindow> windowed = input.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5)));

@PublicEvolving
public <W extends Window> AllWindowedStream<T, W> windowAll(WindowAssigner<? super T, W> assigner) {return new AllWindowedStream(this, assigner);}


@Public
public class AllWindowedStream<T, W extends Window> {private final KeyedStream<T, Byte> input;private final WindowAssigner<? super T, W> windowAssigner;private Trigger<? super T, ? super W> trigger;private Evictor<? super T, ? super W> evictor;private long allowedLateness = 0L;private OutputTag<T> lateDataOutputTag;@PublicEvolvingpublic AllWindowedStream(DataStream<T> input, WindowAssigner<? super T, W> windowAssigner) {//这里设置input的KeySelector为null的对象this.input = input.keyBy(new NullByteKeySelector());this.windowAssigner = windowAssigner;this.trigger = windowAssigner.getDefaultTrigger(input.getExecutionEnvironment());}
}

AllWindowedStream 是对整个数据流应用窗口操作的抽象，而不进行键分组。换句话说，AllWindowedStream 是对全局数据流进行窗口操作。

使用场景:

  1. 当你不需要对数据流进行键分组，而是希望对整个数据流应用窗口操作时，使用 AllWindowedStream。
  2. 适用于全局统计、全局聚合等场景。

2、创建按键分流后的窗口

下面是根据第一位字段当键分流，针对键分的流数据，分别创建窗口

KeyedStream<Tuple2<String, Integer>, String> keyed = input.keyBy(value -> value.f0);
WindowedStream<Tuple2<String, Integer>, String, TimeWindow> windowed = keyed.window(TumblingProcessingTimeWindows.of(Time.seconds(5)));

 @PublicEvolvingpublic <W extends Window> WindowedStream<T, KEY, W> window(WindowAssigner<? super T, W> assigner) {return new WindowedStream(this, assigner);}

@Public
public class WindowedStream<T, K, W extends Window> {private final KeyedStream<T, K> input;//WindowOperatorBuilder 是 Flink 内部用于构建窗口操作符的工具类。它主要用于在内部构建和配置窗口操作符（WindowOperator），并不直接用于用户代码中。WindowOperatorBuilder 提供了一种灵活的方式来配置窗口操作符的各种细节，包括窗口分配器、窗口触发器、窗口合并器等。private final WindowOperatorBuilder<T, K, W> builder;@PublicEvolvingpublic WindowedStream(KeyedStream<T, K> input, WindowAssigner<? super T, W> windowAssigner) {//这里只需要设置input，input的keyBy已经在前面设置了this.input = input;//通过input.getKeySelector()获取KeyedStream设置的函数this.builder = new WindowOperatorBuilder(windowAssigner, windowAssigner.getDefaultTrigger(input.getExecutionEnvironment()), input.getExecutionConfig(), input.getType(), input.getKeySelector(), input.getKeyType());}//调用WindowedStream的trigger实际上调用的是WindowOperatorBuilder的trigger方法@PublicEvolvingpublic WindowedStream<T, K, W> trigger(Trigger<? super T, ? super W> trigger) {this.builder.trigger(trigger);return this;}}   

public class WindowOperatorBuilder<T, K, W extends Window> {private static final String WINDOW_STATE_NAME = "window-contents";private final ExecutionConfig config;private final WindowAssigner<? super T, W> windowAssigner;private final TypeInformation<T> inputType;private final KeySelector<T, K> keySelector;private final TypeInformation<K> keyType;private Trigger<? super T, ? super W> trigger;@Nullableprivate Evictor<? super T, ? super W> evictor;private long allowedLateness = 0L;@Nullableprivate OutputTag<T> lateDataOutputTag;public WindowOperatorBuilder(WindowAssigner<? super T, W> windowAssigner, Trigger<? super T, ? super W> trigger, ExecutionConfig config, TypeInformation<T> inputType, KeySelector<T, K> keySelector, TypeInformation<K> keyType) {this.windowAssigner = windowAssigner;this.config = config;this.inputType = inputType;//把KeyedStream中的keySelector赋值到WindowOperatorBuilder的keySelectorthis.keySelector = keySelector;this.keyType = keyType;this.trigger = trigger;}
}

WindowedStream 是在对数据流进行键分组后，对每个键的子流应用窗口操作的抽象。也就是说，WindowedStream 是对每个键进行独立的窗口操作。

使用场景:

  1. 当你需要对数据流按键分组，并对每个键的子流应用窗口操作时，使用 WindowedStream。
  2. 适用于需要对不同键进行独立统计和聚合的场景。

二、在使用窗口处理数据流时，不同窗口创建的都是窗口算子WindowOperator

这里以聚合函数为例，看不同的窗口类型创建的算子是什么。

1、聚合函数实现

 // 定义聚合函数AggregateFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple2<String, Integer>> aggregateFunction =new AggregateFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple2<String, Integer>>() {@Overridepublic Tuple2<String, Integer> createAccumulator() {return new Tuple2<>("", 0);}@Overridepublic Tuple2<String, Integer> add(Tuple2<String, Integer> value, Tuple2<String, Integer> accumulator) {return new Tuple2<>(value.f0, value.f1 + accumulator.f1);}@Overridepublic Tuple2<String, Integer> getResult(Tuple2<String, Integer> accumulator) {return accumulator;}@Overridepublic Tuple2<String, Integer> merge(Tuple2<String, Integer> a, Tuple2<String, Integer> b) {return new Tuple2<>(a.f0, a.f1 + b.f1);}};//聚合函数接口 public interface AggregateFunction<IN, ACC, OUT> extends Function, Serializable {ACC createAccumulator();ACC add(IN var1, ACC var2);OUT getResult(ACC var1);ACC merge(ACC var1, ACC var2);
}

2、创建全局窗口(入参传的是NullByteKeySelector)

根据上面知道，此时

@Public
public class AllWindowedStream<T, W extends Window> {@PublicEvolvingpublic AllWindowedStream(DataStream<T> input, WindowAssigner<? super T, W> windowAssigner) {//这里设置input的KeySelector为null的对象this.input = input.keyBy(new NullByteKeySelector());this.windowAssigner = windowAssigner;this.trigger = windowAssigner.getDefaultTrigger(input.getExecutionEnvironment());}@PublicEvolvingpublic <ACC, V, R> SingleOutputStreamOperator<R> aggregate(AggregateFunction<T, ACC, V> aggregateFunction, AllWindowFunction<V, R, W> windowFunction, TypeInformation<ACC> accumulatorType, TypeInformation<R> resultType) {//根据AllWindowedStream的构造函数，知道此时this.input.getKeySelector()=new NullByteKeySelectorKeySelector<T, Byte> keySel = this.input.getKeySelector();//省略干扰代码AggregatingStateDescriptor<T, ACC, V> stateDesc = new AggregatingStateDescriptor("window-contents", aggregateFunction, accumulatorType.createSerializer(this.getExecutionEnvironment().getConfig()));operator = new WindowOperator(this.windowAssigner, this.windowAssigner.getWindowSerializer(this.getExecutionEnvironment().getConfig()), keySel, this.input.getKeyType().createSerializer(this.getExecutionEnvironment().getConfig()), stateDesc, new InternalSingleValueAllWindowFunction(windowFunction), this.trigger, this.allowedLateness, this.lateDataOutputTag);//省略干扰代码   return this.input.transform(opName, resultType, (OneInputStreamOperator)operator).forceNonParallel();}}
@PublicEvolving
public class AggregatingStateDescriptor<IN, ACC, OUT> extends StateDescriptor<AggregatingState<IN, OUT>, ACC> {private final AggregateFunction<IN, ACC, OUT> aggFunction;public AggregatingStateDescriptor(String name, AggregateFunction<IN, ACC, OUT> aggFunction, TypeSerializer<ACC> typeSerializer) {super(name, typeSerializer, (Object)null);this.aggFunction = (AggregateFunction)Preconditions.checkNotNull(aggFunction);}
}

3、创建按键分流后的窗口(入参传的是KeyedStream的KeySelector)

public class WindowedStream<T, K, W extends Window> {@PublicEvolvingpublic WindowedStream(KeyedStream<T, K> input, WindowAssigner<? super T, W> windowAssigner) {this.input = input;this.builder = new WindowOperatorBuilder(windowAssigner, windowAssigner.getDefaultTrigger(input.getExecutionEnvironment()), input.getExecutionConfig(), input.getType(), input.getKeySelector(), input.getKeyType());}public <ACC, V, R> SingleOutputStreamOperator<R> aggregate(AggregateFunction<T, ACC, V> aggregateFunction, WindowFunction<V, R, K, W> windowFunction, TypeInformation<ACC> accumulatorType, TypeInformation<R> resultType) {//删除干扰代码aggregateFunction = (AggregateFunction)this.input.getExecutionEnvironment().clean(aggregateFunction);String opName = this.builder.generateOperatorName(aggregateFunction, windowFunction);OneInputStreamOperator<T, R> operator = this.builder.aggregate(aggregateFunction, windowFunction, accumulatorType);return this.input.transform(opName, resultType, operator);}}

通过上面我们知道builder指的是WindowOperatorBuilder，并且构造函数入参中的keySelector实际上是keyedStream的keySelector

public class WindowOperatorBuilder<T, K, W extends Window> {public WindowOperatorBuilder(WindowAssigner<? super T, W> windowAssigner, Trigger<? super T, ? super W> trigger, ExecutionConfig config, TypeInformation<T> inputType, KeySelector<T, K> keySelector, TypeInformation<K> keyType) {this.windowAssigner = windowAssigner;this.config = config;this.inputType = inputType;//这个keySelector = keyedStream的keySelectorthis.keySelector = keySelector;this.keyType = keyType;this.trigger = trigger;}public <ACC, V, R> WindowOperator<K, T, ?, R, W> aggregate(AggregateFunction<T, ACC, V> aggregateFunction, WindowFunction<V, R, K, W> windowFunction, TypeInformation<ACC> accumulatorType) {//删除干扰代码AggregatingStateDescriptor<T, ACC, V> stateDesc = new AggregatingStateDescriptor("window-contents", aggregateFunction, accumulatorType.createSerializer(this.config));return this.buildWindowOperator(stateDesc, new InternalSingleValueWindowFunction(windowFunction));}private <ACC, R> WindowOperator<K, T, ACC, R, W> buildWindowOperator(StateDescriptor<? extends AppendingState<T, ACC>, ?> stateDesc, InternalWindowFunction<ACC, R, K, W> function) {return new WindowOperator(this.windowAssigner, this.windowAssigner.getWindowSerializer(this.config), this.keySelector, this.keyType.createSerializer(this.config), stateDesc, function, this.trigger, this.allowedLateness, this.lateDataOutputTag);}}  
}

两种窗口最后都是构建WindowOperator，只是传的参数不一样，其中全局窗口的keySelector是null对象，按键建窗口的keySelector是取的KeyedStream的

3、WindowOperator

@Internal
public class WindowOperator<K, IN, ACC, OUT, W extends Window> extends AbstractUdfStreamOperator<OUT, InternalWindowFunction<ACC, OUT, K, W>> implements OneInputStreamOperator<IN, OUT>, Triggerable<K, W> {private final KeySelector<IN, K> keySelector;private transient InternalAppendingState<K, W, IN, ACC, ACC> windowState;private final StateDescriptor<? extends AppendingState<IN, ACC>, ?> windowStateDescriptor;public WindowOperator(WindowAssigner<? super IN, W> windowAssigner, TypeSerializer<W> windowSerializer, KeySelector<IN, K> keySelector, TypeSerializer<K> keySerializer, StateDescriptor<? extends AppendingState<IN, ACC>, ?> windowStateDescriptor, InternalWindowFunction<ACC, OUT, K, W> windowFunction, Trigger<? super IN, ? super W> trigger, long allowedLateness, OutputTag<IN> lateDataOutputTag) {//删除干扰代码    this.windowStateDescriptor = windowStateDescriptor;this.keySelector = (KeySelector)Preconditions.checkNotNull(keySelector);}public void open() throws Exception {if (this.windowStateDescriptor != null) {this.windowState = (InternalAppendingState)this.getOrCreateKeyedState(this.windowSerializer, this.windowStateDescriptor);}}//数据到的执行方法public void processElement(StreamRecord<IN> element) throws Exception {//它遍历名为elementWindows的迭代器Collection<W> elementWindows = this.windowAssigner.assignWindows(element.getValue(), element.getTimestamp(), this.windowAssignerContext);//这里有判断窗口是否是像会话窗口那种需要动态合并窗口的逻辑，为了不干扰理解，这里删除了那一块代码逻辑，有兴趣的可以专门去看一下//删除干扰代码Iterator var12 = elementWindows.iterator();label59:while(true) {Window window;TriggerResult triggerResult;while(true) {//在每次迭代中，它会检查窗口是否已经过期（isWindowLate方法）do {if (!var12.hasNext()) {break label59;}window = (Window)var12.next();} while(this.isWindowLate(window));//更新窗口的状态，将元素值添加到窗口状态中，并在触发器上下文中设置键和窗口isSkippedElement = false;this.windowState.setCurrentNamespace(window);//add方法this.windowState.add(element.getValue());this.triggerContext.key = key;this.triggerContext.window = window;//调用onElement方法对元素进行处理并检查触发器结果triggerResult = this.triggerContext.onElement(element);if (!triggerResult.isFire()) {//如果触发结果不需要触发（isFire() 返回 false），则跳出内部循环。break;}//如果窗口内容不为空，它将发出窗口内容并终止内部循环ACC contents = this.windowState.get();if (contents != null) {this.emitWindowContents(window, contents);break;}}//如果触发器结果要求清除窗口（isPurge()返回true），则会清除窗口状态if (triggerResult.isPurge()) {this.windowState.clear();}this.registerCleanupTimer(window);}}//水位线判断逻辑protected boolean isWindowLate(W window) {return this.windowAssigner.isEventTime() && this.cleanupTime(window) <= this.internalTimerService.currentWatermark();}}

这里又发现了熟悉的接口，OneInputStreamOperator<IN, OUT>，processElement方法实际上是父类接口Input<IN>的processElement方法

下面是WindowOperator的类关系图，和Flink 1.14.*中flatMap，filter等基本转换函数源码中RichFlatMapFunction和RichFilterFunction一样的父类AbstractUdfStreamOperator ，接口新增了特性
在这里插入图片描述
通过这些，大家心里应该有数了，不管是FlatMap还是Filter还是窗口，都是基于这个类关系图扩展下来的

这篇关于Flink 1.14.* Flink窗口创建和窗口计算源码的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/1119893。

相关文章

Android 悬浮窗开发示例((动态权限请求 | 前台服务和通知 | 悬浮窗创建 )

Android 悬浮窗开发示例((动态权限请求 | 前台服务和通知 | 悬浮窗创建 )

《Android悬浮窗开发示例((动态权限请求|前台服务和通知|悬浮窗创建)》本文介绍了Android悬浮窗的实现效果,包括动态权限请求、前台服务和通知的使用,悬浮窗权限需要动态申请并引导... 目录一、悬浮窗 动态权限请求1、动态请求权限2、悬浮窗权限说明3、检查动态权限4、申请动态权限5、权限设置完毕后
阅读更多...
Python如何计算两个不同类型列表的相似度

Python如何计算两个不同类型列表的相似度

《Python如何计算两个不同类型列表的相似度》在编程中,经常需要比较两个列表的相似度,尤其是当这两个列表包含不同类型的元素时,下面小编就来讲讲如何使用Python计算两个不同类型列表的相似度吧... 目录摘要引言数字类型相似度欧几里得距离曼哈顿距离字符串类型相似度Levenshtein距离Jaccard相
阅读更多...
Python创建Excel的4种方式小结

Python创建Excel的4种方式小结

《Python创建Excel的4种方式小结》这篇文章主要为大家详细介绍了Python中创建Excel的4种常见方式,文中的示例代码简洁易懂,具有一定的参考价值,感兴趣的小伙伴可以学习一下... 目录库的安装代码1——pandas代码2——openpyxl代码3——xlsxwriterwww.cppcns.c
阅读更多...
使用Python在Excel中创建和取消数据分组

使用Python在Excel中创建和取消数据分组

《使用Python在Excel中创建和取消数据分组》Excel中的分组是一种通过添加层级结构将相邻行或列组织在一起的功能,当分组完成后,用户可以通过折叠或展开数据组来简化数据视图,这篇博客将介绍如何使... 目录引言使用工具python在Excel中创建行和列分组Python在Excel中创建嵌套分组Pyt
阅读更多...
解决IDEA使用springBoot创建项目,lombok标注实体类后编译无报错,但是运行时报错问题

解决IDEA使用springBoot创建项目,lombok标注实体类后编译无报错,但是运行时报错问题

《解决IDEA使用springBoot创建项目,lombok标注实体类后编译无报错,但是运行时报错问题》文章详细描述了在使用lombok的@Data注解标注实体类时遇到编译无误但运行时报错的问题,分析... 目录问题分析问题解决方案步骤一步骤二步骤三总结问题使用lombok注解@Data标注实体类，编译时
阅读更多...
MySQL分表自动化创建的实现方案

MySQL分表自动化创建的实现方案

《MySQL分表自动化创建的实现方案》在数据库应用场景中,随着数据量的不断增长,单表存储数据可能会面临性能瓶颈,例如查询、插入、更新等操作的效率会逐渐降低,分表是一种有效的优化策略,它将数据分散存储在... 目录一、项目目的二、实现过程（一）mysql 事件调度器结合存储过程方式1. 开启事件调度器2. 创
阅读更多...
Go中sync.Once源码的深度讲解

Go中sync.Once源码的深度讲解

《Go中sync.Once源码的深度讲解》sync.Once是Go语言标准库中的一个同步原语,用于确保某个操作只执行一次,本文将从源码出发为大家详细介绍一下sync.Once的具体使用,x希望对大家有... 目录概念简单示例源码解读总结概念sync.Once是Go语言标准库中的一个同步原语，用于确保某个操
阅读更多...
mysql外键创建不成功/失效如何处理

mysql外键创建不成功/失效如何处理

《mysql外键创建不成功/失效如何处理》文章介绍了在MySQL5.5.40版本中,创建带有外键约束的`stu`和`grade`表时遇到的问题,发现`grade`表的`id`字段没有随着`studen... 当前mysql版本：SELECT VERSION();结果为：5.5.40。在复习mysql外键约
阅读更多...
Window Server创建2台服务器的故障转移群集的图文教程

Window Server创建2台服务器的故障转移群集的图文教程

《WindowServer创建2台服务器的故障转移群集的图文教程》本文主要介绍了在WindowsServer系统上创建一个包含两台成员服务器的故障转移群集,文中通过图文示例介绍的非常详细,对大家的... 目录一、 准备条件二、在ServerB安装故障转移群集三、在ServerC安装故障转移群集，操作与Ser
阅读更多...
Window Server2016 AD域的创建的方法步骤

Window Server2016 AD域的创建的方法步骤

《WindowServer2016AD域的创建的方法步骤》本文主要介绍了WindowServer2016AD域的创建的方法步骤,文中通过图文介绍的非常详细,对大家的学习或者工作具有一定的参考学习价... 目录一、准备条件二、在ServerA服务器中常见AD域管理器：三、创建AD域，域地址为“test.ly”
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2