Flink处理函数实战之一:深入了解ProcessFunction的状态(Flink-1.10)

本文涉及的产品
实时计算 Flink 版,1000CU*H 3个月
简介: 深入了解ProcessFunction的状态操作

欢迎访问我的GitHub

这里分类和汇总了欣宸的全部原创(含配套源码):https://github.com/zq2599/blog_demos

Flink处理函数实战系列链接

  1. 深入了解ProcessFunction的状态操作(Flink-1.10)
  2. ProcessFunction
  3. KeyedProcessFunction类
  4. ProcessAllWindowFunction(窗口处理)
  5. CoProcessFunction(双流处理)

关于ProcessFunction状态的疑惑

学习Flink的ProcessFunction过程中,官方文档中涉及状态处理的时候,不止一次提到只适用于keyed stream的元素,如下图红框所示:
在这里插入图片描述
之前写过一些flink应用,keyed stream常用但不是必须用的,所以产生了疑问:

  1. 为何只有keyed stream的元素能读写状态?
  2. 每个key对应的状态是如何操作的?

    Flink的"状态"

    先去回顾Flink"状态"的知识点:
  3. 官方文档说就两种状态:keyed state和operator state:
    在这里插入图片描述
  4. 如上图,keyed stream的元素是具有key的特征,与ProcessFunction的操作状态时要求匹配,其他steam的元素由于没有key的特征,所以也就没有状态一说了;
  5. 另一种状态是Operator State,如下图,这是和多并行度计算时的算子实例绑定的,例如当前算子消费kafka的某个分区的最新offset,而ProcessFunction是用来处理stream元素的,不会涉及到Operator State:
    在这里插入图片描述

    官方demo

    为了学习ProcessFunction就去看官方demo,地址是:https://ci.apache.org/projects/flink/flink-docs-release-1.10/dev/stream/operators/process_function.html ,简单说说这个demo的功能:
  6. 数据源在不间断的产生单词,每个单词对应一个Tuple2的实例;
  7. 数据源被keyBy方法转成KeyedStream,key是Tuple2实例的f0字段;
  8. 一个KeyedProcessFunction的子类CountWithTimeoutFunction,被用来处理KeyedStream的每个元素,处理的逻辑:为每个key维护一个状态,状态的内容是这个key的出现次数和最后一次出现时间;
  9. 如果那个key连续一分钟没有出现,KeyedProcessFunction就向下游发送这个元素;

以上就是官方demo的功能,本来是想通过demo来加深认识,结果看完不但没有明白,反而更晕了,下图是我对demo代码的疑惑:
在这里插入图片描述
从上图可见我的疑惑,这里再复述一下:

  1. 入参value是Tuple2类型,假设其f0字段等于aaa,那么processElement方法的作用,就是取出aaa的状态,更新后保存;
  2. 从代码上看,state.value()返回了aaa的状态,这个value方法并没有将aaa作为入参,那怎么做到返回aaa的状态呢?如果下一个入参value的f0字段等于bbb了,这个state.value()能返回bbb的状态吗?
  3. 对更新状态的代码state.update(current)也是同样的疑惑;
  4. 然后又产生了新的疑惑:成员变量state难道是一直在变?每执行一次processElement,都会变成该key对应的state实例?

    先反思为何会有上述疑惑

  5. 上述疑惑产生的原因,应该是受到平时使用HashMap的影响,HashMap获取值就是在调用get方法时指定key,设置值也是在put时指定key,所以看到state.value()方法没有用key做入参就不习惯了
  6. 要消除这种不适应,要做的第一件事就是提醒自己:processElement是在框架内运行的,很多数据在之前已经由框架准备好了;
  7. 接下来要做的,就是把框架准备数据的逻辑看一遍,除了弄明白自己的问题,由于ProcessFunction属于最低阶抽象(如下图的最下方位置),看懂了这些,其实也是在了解DataStream/DataSet API的设计思路:
    在这里插入图片描述

    跟踪源码

  8. 如下图,让我们从一个断点的堆栈开始吧,这是在执行上面demo中的processElement方法之前的一个断点,可见根源是个线程的run方法,也就是KeyedProcessFunction对应的算子执行任务的线程:
    在这里插入图片描述
  9. 上面的堆栈不必每一层都细看,只关注重要的部分,下图这段很重要:StreamTask.run方法中,有个无限循环(猜测是每次执行processInput方法都处理KeyedStream的一个元素):
    在这里插入图片描述
  10. 如下图,StreamOneInputProcessor.processInput方法取出KeyedStream的一个元素,调用processElement方法,并将此元素作为入参,再结合上一幅图可以看出:在编写KeyedProcessFunction子类的时候,KeyedStream的每个元素都会作为入参,在调用你重写的processElement方法时传进去;这一点,在做ProcessFunction和KeyedProcessFunction开发时都是要格外注意的:
    在这里插入图片描述
  11. 接下来到了最关键的地方了,下图红框中的streamOperator.setKeyContextElement1(record)会解答我前面的疑惑,一定要进去看个清楚,(后面的黄线上的代码,您应该猜到了,里面其实就是调用demo中的processElement方法)
    在这里插入图片描述
  12. 下图中,AbstractStreamOperator.setKeyContextElement给出了答案:对于KeyedStream的每个元素,都会在这里算出key,再调用setCurrentKey保存这个key
    在这里插入图片描述
  13. 展开setCurrentKey,如下图,发现key的保存和当前状态的存储策略(StateBackend)有关,我这里是默认策略HeapKeyedStateBackend
    在这里插入图片描述
  14. 最终,根据当前元素得到的key会在StateBackend的keyContext对象中找地方保存,StateBackend的具体实现和Flink设置有关,我这里是保存到了InternalKeyContextImpl实例的currentKey变量中:
    在这里插入图片描述
  15. 代码读到这里,对我前面的疑惑,您应该能推测出答案了:state.value()里面会通过StateBackend的keyContext取出刚才保存的key,接下来就能像HashMap那样根据key查出该key的状态了,接下来是愉快的印证我们推测的过程;
  16. state.value()代码位置打断点一次看个明白,如下图,果然,state里面有StateBackend的keyContext对象的引用,访问刚才保存的key就不成问题了:
    在这里插入图片描述
  17. 展开state.value()方法如下,简单明了,直接拿keyContext保存的key作为入参去取对应的状态:
    在这里插入图片描述
  18. 再展开上面的get方法,可见最终是从stateMap中取得的,而这个stateMap的具体实现是CopyOnWriteStateMap类型的实例:
    在这里插入图片描述
  19. 代码读到这里,只剩最后一处需要印证了:更新状态的state.update(current)方法,应该也是以StateBackend的keyContext中的key作为自己的key,再将入参的current作为value,更新到stateMap中,来吧,一起印证这个推测;
  20. 展开方法,看到的是stateTable.put方法(前面刚看过stateTable的get方法,稳了):
    在这里插入图片描述
  21. stateTable.put方法里面和前面的get方法一样,直接拿keyContext保存的key作为自己的key:
    在这里插入图片描述
  22. 最终是调用了stateMap.put方法,将数据保存在CopyOnWriteStateMap实例中:
    在这里插入图片描述
  23. 得益于Flink代码自身规范、清晰的设计和实现,再加上IDEA强大的debug功能,整个阅读和分析过程十分顺利,这其中的收获会逐渐在今后深入学习DataStreamAPI的过程中见效;

最后,根据上面的分析过程绘制了一幅简陋的流程图,希望能帮助您加快理解:
在这里插入图片描述

欢迎关注阿里云开发者社区:程序员欣宸

学习路上,你不孤单,欣宸原创一路相伴...

相关实践学习
基于Hologres+Flink搭建GitHub实时数据大屏
通过使用Flink、Hologres构建实时数仓,并通过Hologres对接BI分析工具(以DataV为例),实现海量数据实时分析.
实时计算 Flink 实战课程
如何使用实时计算 Flink 搞定数据处理难题?实时计算 Flink 极客训练营产品、技术专家齐上阵,从开源 Flink功能介绍到实时计算 Flink 优势详解,现场实操,5天即可上手! 欢迎开通实时计算 Flink 版: https://cn.aliyun.com/product/bigdata/sc Flink Forward Asia 介绍: Flink Forward 是由 Apache 官方授权,Apache Flink Community China 支持的会议,通过参会不仅可以了解到 Flink 社区的最新动态和发展计划,还可以了解到国内外一线大厂围绕 Flink 生态的生产实践经验,是 Flink 开发者和使用者不可错过的盛会。 去年经过品牌升级后的 Flink Forward Asia 吸引了超过2000人线下参与,一举成为国内最大的 Apache 顶级项目会议。结合2020年的特殊情况,Flink Forward Asia 2020 将在12月26日以线上峰会的形式与大家见面。
相关文章
|
4月前
|
消息中间件 运维 Kafka
直播预告|Kafka+Flink双引擎实战:手把手带你搭建分布式实时分析平台!
在数字化转型中,企业亟需从海量数据中快速提取价值并转化为业务增长动力。5月15日19:00-21:00,阿里云三位技术专家将讲解Kafka与Flink的强强联合方案,帮助企业零门槛构建分布式实时分析平台。此组合广泛应用于实时风控、用户行为追踪等场景,具备高吞吐、弹性扩缩容及亚秒级响应优势。直播适合初学者、开发者和数据工程师,参与还有机会领取定制好礼!扫描海报二维码或点击链接预约直播:[https://developer.aliyun.com/live/255088](https://developer.aliyun.com/live/255088)
340 35
直播预告|Kafka+Flink双引擎实战:手把手带你搭建分布式实时分析平台!
|
4月前
|
消息中间件 运维 Kafka
直播预告|Kafka+Flink 双引擎实战:手把手带你搭建分布式实时分析平台!
直播预告|Kafka+Flink 双引擎实战:手把手带你搭建分布式实时分析平台!
168 11
|
10月前
|
分布式计算 大数据 Apache
ClickHouse与大数据生态集成:Spark & Flink 实战
【10月更文挑战第26天】在当今这个数据爆炸的时代,能够高效地处理和分析海量数据成为了企业和组织提升竞争力的关键。作为一款高性能的列式数据库系统,ClickHouse 在大数据分析领域展现出了卓越的能力。然而,为了充分利用ClickHouse的优势,将其与现有的大数据处理框架(如Apache Spark和Apache Flink)进行集成变得尤为重要。本文将从我个人的角度出发,探讨如何通过这些技术的结合,实现对大规模数据的实时处理和分析。
698 2
ClickHouse与大数据生态集成:Spark & Flink 实战
|
6月前
|
消息中间件 JSON 数据库
探索Flink动态CEP:杭州银行的实战案例
探索Flink动态CEP:杭州银行的实战案例
210 5
|
9月前
|
消息中间件 JSON 数据库
探索Flink动态CEP:杭州银行的实战案例
本文由杭州银行大数据工程师唐占峰、欧阳武林撰写,介绍Flink动态CEP的定义、应用场景、技术实现及使用方式。Flink动态CEP是基于Flink的复杂事件处理库,支持在不重启服务的情况下动态更新规则,适应快速变化的业务需求。文章详细阐述了其在反洗钱、反欺诈和实时营销等金融领域的应用,并展示了某金融机构的实际应用案例。通过动态CEP,用户可以实时调整规则,提高系统的灵活性和响应速度,降低维护成本。文中还提供了具体的代码示例和技术细节,帮助读者理解和使用Flink动态CEP。
1106 2
探索Flink动态CEP:杭州银行的实战案例
|
9月前
|
数据处理 数据安全/隐私保护 流计算
Flink 三种时间窗口、窗口处理函数使用及案例
Flink 是处理无界数据流的强大工具,提供了丰富的窗口机制。本文介绍了三种时间窗口(滚动窗口、滑动窗口和会话窗口)及其使用方法,包括时间窗口的概念、窗口处理函数的使用和实际案例。通过这些机制,可以灵活地对数据流进行分析和计算,满足不同的业务需求。
1006 27
|
11月前
|
SQL 消息中间件 分布式计算
大数据-115 - Flink DataStream Transformation 多个函数方法 FlatMap Window Aggregations Reduce
大数据-115 - Flink DataStream Transformation 多个函数方法 FlatMap Window Aggregations Reduce
132 0
|
1月前
|
存储 分布式计算 数据处理
「48小时极速反馈」阿里云实时计算Flink广招天下英雄
阿里云实时计算Flink团队,全球领先的流计算引擎缔造者,支撑双11万亿级数据处理,推动Apache Flink技术发展。现招募Flink执行引擎、存储引擎、数据通道、平台管控及产品经理人才,地点覆盖北京、杭州、上海。技术深度参与开源核心,打造企业级实时计算解决方案,助力全球企业实现毫秒洞察。
364 0
「48小时极速反馈」阿里云实时计算Flink广招天下英雄
|
运维 数据处理 数据安全/隐私保护
阿里云实时计算Flink版测评报告
该测评报告详细介绍了阿里云实时计算Flink版在用户行为分析与标签画像中的应用实践,展示了其毫秒级的数据处理能力和高效的开发流程。报告还全面评测了该服务在稳定性、性能、开发运维及安全性方面的卓越表现,并对比自建Flink集群的优势。最后,报告评估了其成本效益,强调了其灵活扩展性和高投资回报率,适合各类实时数据处理需求。

热门文章

最新文章