StreamOperator源码简析

朴灿烈づ我的快乐病毒、 2023-06-19 02:25 29阅读 0赞

点击上方蓝

字关注~

StreamOperator是任务执行过程中实际处理类，上层由StreamTask调用，下层调用UserFunction,列举一些常见的StreamOperator

*  env.addSource对应StreamSource
 *  dataStream.map 对应StreamMap
 *  dataStrem.window对应WindowOperator
 *  dataStream.addSink对应StreamSink
 *  dataStream.keyBy(..).process对应KeyedProcessOperator

StreamOperator涉及数据处理、checkpoint、状态存储、定时调用等，本篇幅将从源码角度分析StreamOperator所涉及的核心调用流程。

## StreamOperator层级结构 ##

![format_png][]

最顶层是一个StreamOperator的接口，定义了其生命周期一些方法，继承接口如下：

*  CheckpointListener接口，notifyCheckpointComplete表示checkpoint完成后的回调方法
 *  KeyContext接口，用于当前key的切换，使用在KeyedStream中state的key设置
 *  Disposable接口，dispose方法定义了资源释放
 *  Serializable序列化接口

AbstractStreamOperator是StreamOperator的基础抽象实现类，所有的operator都必须继承该抽象类；  
AbstractUdfStreamOperator 是继承AbstractStreamOperator的抽象实现类，其内部包含了userFunction, 在Task的生命周期都会调用userFunction中对应的方法；  
OneInputStreamOperator、TwoInputStreamOperator都是继承StreamOperator的接口，分别表示处理一个输入、两个输入的Operator，包含了processElement/processWatermark/processLatencyMarker方法;

*  OneInputStreamOperator实现类StreamMap、WindowOperator、KeyedProcessOperator等单流入处理operator
 *  TwoInputStreamOperator实现类CoStreamMap、KeyedCoProcessOperator、IntervalJoinOperator等多流处理operator

StreamSource表示的source端的operator,其既没有实现OneInputStreamOperator接口也没有实现TwoInputStreamOperator接口，由于其为流处理的源头，不需要接受输入

## AbstractStreamOperator/AbstractUdfStreamOperator分析 ##

AbstractStreamOperator是所有operator的基础抽象类，而AbstractUdfStreamOperator则是面向userFunction调用，接下来分析一下这两个类，其大部分方法都是由StreamTask触发调用，用于初始化或者资源释放等操作,以StreamTask.invoke方法为入口来分析里面的方法：

*  initializeState状态初始化，会调用到StreamOperator的initializeState方法，初始化operatorStateBackend/keyedStateBackend状态后端，定时器恢复初始化，对于KeyedState来说会自动初始化恢复，但是operatorState则需要手动初始化恢复，所以在其继承的AbstractUdfStreamOperator会调用userFunction的initializeState方法，前提是该userFunction需要实现CheckpointedFunction接口；
 *  open初始化方法，在AbstractStreamOperator中是一个空的实现，通常可以在userFunction重写open方法完成一些用户初始化工作，例如创建资源链接
 *  run方法，在任务正常情况下一直执行的方法，根据收到的不同数据类型调用AbstractStreamOperator不同的方法

1.  如果是watermark,会调用其processWatermark方法，在该方法里面做一些定时触发的判断与调用
2.  如果是LatencyMarker，其表示的是一个延时标记，同于统计数据从source到下游operator耗时，会调用 processLatencyMarker方法，在该方法里面会上报Histogram类型的metric, 在默认情况下该功能是关闭的
3.  如果是StreamRecord，也就是处理的业务数据，首先会调用setKeyContextElement方法，用于切换 KeyedStream类型的的statebackend的当前key, 然后调用processElement具体的数据处理流程
4.  如果是CheckpointBarrier，表示的是需要checkpoint，首先会调用prepareSnapshotPreBarrier方法，在AbstractStreamOperator中是一个空实现doNothing,然后调用snapshotState方法，在AbstractUdfStreamOperator会调用userFunction的snapshotState方法，前提是该userFunction需要实现CheckpointedFunction接口；

*  close方法，任务正常结束调用方法，在AbstractStreamOperator中是一个空的实现，通常可以在userFunction重写close方法完成一些资源释放；
 *  dispose方法，任务正常结束或者异常结束调用的方法，如果是异常结束那么就会调用到close方法，正常结束不会重复调用，在dispose里面完成一些状态最终资源的释放;

其他方法：

*  setup方法，初始化做一些参数配置
 *  notifyCheckpointComplete方法，在checkpoint完成时调用的方法，面向用户实现的userFunction需要实现CheckpointListener接口

推荐阅读

[1. Flink中延时调用设计与实现][1._Flink]

[2. Flink维表关联系列之Hbase维表关联：LRU策略][2. Flink_Hbase_LRU]

[3. 你应该了解的Watermark][3._Watermark]

[4. Flink exactly-once系列之事务性输出实现][4._Flink exactly-once]

[5. Flink时间系统系列之实例讲解：如何做定时输出][5. Flink]

[6. Flink实战:全局TopN分析与实现][6._Flink_TopN]

[7. Flink per-Job模式InfluxdbReporter上报JobName][7._Flink per-Job_InfluxdbReporter_JobName]

[8. Flink SQL自定义聚合函数][8._Flink SQL]

关注回复Flink获取更多信息~

![format_png 1][]

[format_png]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9tbWJpei5xcGljLmNuL21tYml6X3BuZy9TN3RJdm81eFlNNm0yd1RPZERQTFFRcHFpY0N5NEpHTnJTS0ZUZUNqZWVHdlRvMlhqRWppYThmWXV5aGo5cnRrVVZrRzJxbW1xZHQ0ZjJNdDJWVDdIR3RBLzY0MA?x-oss-process=image/format,png
[1._Flink]: http://mp.weixin.qq.com/s?__biz=MzU5MTc1NDUyOA%3D%3D&chksm=fe2b6668c95cef7e3639c575ac8cb899e7317f82c2b446a23335c8f33dc97a968ea7f345b914&idx=1&mid=2247483943&scene=21&sn=d3e1002255cbd6cfe16855639f82b80f#wechat_redirect
[2. Flink_Hbase_LRU]: http://mp.weixin.qq.com/s?__biz=MzU5MTc1NDUyOA%3D%3D&chksm=fe2b6640c95cef5633a4ea18b81fbe2f0ecded5f9004226a3dfec6e75c094b99c58ffbf28307&idx=1&mid=2247483919&scene=21&sn=23628a87a5de9fec669248664df86e78#wechat_redirect
[3._Watermark]: http://mp.weixin.qq.com/s?__biz=MzU5MTc1NDUyOA%3D%3D&chksm=fe2b6645c95cef533c8044be2f06be988e08449ac47432e1370fc5c6d7aac0ebf1c82fdca84c&idx=1&mid=2247483914&scene=21&sn=c2d18ea26669adb69df9fdb1845751c5#wechat_redirect
[4._Flink exactly-once]: http://mp.weixin.qq.com/s?__biz=MzU5MTc1NDUyOA%3D%3D&chksm=fe2b6598c95cec8eca29fc8c23b789bf1b11e250188ce3de0d0e462d0ad463ad94042b408990&idx=1&mid=2247483863&scene=21&sn=4dfc5b69f6488985aee4086f0fa9d372#wechat_redirect
[5. Flink]: http://mp.weixin.qq.com/s?__biz=MzU5MTc1NDUyOA%3D%3D&chksm=fe2b65d5c95cecc39bf4fd92b2e5b798a52be30b0359fe95d84c8049ee952439831de13e179b&idx=1&mid=2247483802&scene=21&sn=0fdb31192bbacd2062468ef7a050b540#wechat_redirect
[6._Flink_TopN]: http://mp.weixin.qq.com/s?__biz=MzU5MTc1NDUyOA%3D%3D&chksm=fe2b662ec95cef38b10fb32240d62c6bc934020b6dd2476a8911e13ff9c128102fa131f59467&idx=1&mid=2247484001&scene=21&sn=e7c0a159023d7552c5354a923e4dbce4#wechat_redirect
[7._Flink per-Job_InfluxdbReporter_JobName]: http://mp.weixin.qq.com/s?__biz=MzU5MTc1NDUyOA%3D%3D&chksm=fe2b6629c95cef3f984993c16b6e4c291493a227a604f8b480afe394f1ee32cc383be3a496f1&idx=1&mid=2247484006&scene=21&sn=688e365f39719d56732e46929dffa97c#wechat_redirect
[8._Flink SQL]: http://mp.weixin.qq.com/s?__biz=MzU5MTc1NDUyOA%3D%3D&chksm=fe2b66c9c95cefdf2d41af1706969e92e053ab8db90e8224cfd969fdb5ad3eeb71c9aabc8ea7&idx=1&mid=2247484038&scene=21&sn=ae0dc2ce5482279c6188379a55a4c7b5#wechat_redirect
[format_png 1]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9tbWJpei5xcGljLmNuL21tYml6X3BuZy9TN3RJdm81eFlNNTdzV2VYeTJpY0xTdHY4MUNRU1ZINUFKa0J4aWFPWWRPRWlhaWNMM1R1akJnSWJHM0ZZaWNoZ0pXaWF2QWhtRXI3OG1BS1hvb3h3dkxUZVNCQS82NDA?x-oss-process=image/format,png