关于云计算:Flink处理函数实战之一深入了解ProcessFunction的状态Flink110

欢送拜访我的GitHub

https://github.com/zq2599/blog_demos

内容：所有原创文章分类汇总及配套源码，波及Java、Docker、Kubernetes、DevOPS等；

欢送拜访我的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. 为何只有<font color=”blue”>keyed stream</font>的元素能读写状态？
2. 每个key对应的状态是如何操作的？

Flink的”状态”

先去回顾Flink”状态”的知识点：

1. 官网文档说就两种状态：keyed state和operator state：

1. 如上图，keyed stream的元素是具备key的特色，与ProcessFunction的操作状态时要求匹配，其余steam的元素因为没有key的特色，所以也就没有<font color=”blue”>状态</font>一说了；
2. 另一种状态是<font color=”blue”>Operator State</font>，如下图，这是和多并行度计算时的算子实例绑定的，例如以后算子生产kafka的某个分区的最新offset，而ProcessFunction是用来解决stream元素的，不会波及到Operator State：

官网demo

为了学习ProcessFunction就去看官网demo，地址是：https://ci.apache.org/project… ，简略说说这个demo的性能：

1. 数据源在不间断的产生单词，每个单词对应一个Tuple2<String,String>的实例；
2. 数据源被<font color=”blue”>keyBy</font>办法转成KeyedStream，key是Tuple2实例的f0字段；
3. 一个KeyedProcessFunction的子类<font color=”blue”>CountWithTimeoutFunction</font>，被用来解决KeyedStream的每个元素，解决的逻辑：为每个key保护一个状态，状态的内容是这个key的呈现次数和最初一次呈现工夫；
4. 如果那个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实例？

先反思为何会有上述纳闷

1. 上述纳闷产生的起因，应该是受到平时应用HashMap的影响，HashMap获取值就是在调用get办法时指定key，设置值也是在put时指定key，所以看到state.value()办法没有用key做入参就不习惯了
2. 要打消这种不适应，要做的第一件事就是揭示本人：processElement是在框架内运行的，很多数据在之前曾经由框架筹备好了；
3. 接下来要做的，就是把<font color=”blue”>框架筹备数据</font>的逻辑看一遍，除了弄明确本人的问题，因为ProcessFunction属于最低阶形象(如下图的最下方地位)，看懂了这些，其实也是在理解DataStream/DataSet API的设计思路：

跟踪源码

1. 如下图，让咱们从一个断点的堆栈开始吧，这是在执行下面demo中<font color=”blue”>的processElement</font>办法之前的一个断点，可见本源是个线程的run办法，也就是KeyedProcessFunction对应的算子执行工作的线程：

1. 下面的堆栈不用每一层都细看，只关注重要的局部，下图这段很重要：StreamTask.run办法中，有个有限循环（猜想是每次执行processInput办法都解决KeyedStream的一个元素）：

1. 如下图，StreamOneInputProcessor.processInput办法取出KeyedStream的一个元素，调用processElement办法，并将此元素作为入参，再联合上一幅图能够看出：在编写<font color=”red”>KeyedProcessFunction子类的时候，KeyedStream的每个元素都会作为入参，在调用你重写的processElement办法时传进去；</font>这一点，在做ProcessFunction和KeyedProcessFunction开发时都是要分外留神的：

1. 接下来到了最要害的中央了，下图红框中的streamOperator.setKeyContextElement1(record)会解答我后面的纳闷，肯定要进去看个分明，(前面的黄线上的代码，您应该猜到了，外面其实就是调用demo中的processElement办法)

1. 下图中，AbstractStreamOperator.setKeyContextElement给出了答案：<font color=”blue”>对于KeyedStream的每个元素，都会在这里算出key，再调用setCurrentKey保留这个key</font>：

1. 开展<font color=”blue”>setCurrentKey</font>，如下图，发现key的保留和以后状态的存储策略(StateBackend)无关，我这里是默认策略<font color=”blue”>HeapKeyedStateBackend</font>：

1. 最终，依据以后元素失去的key会在StateBackend的keyContext对象中找中央保留，StateBackend的具体实现和Flink设置无关，我这里是保留到了InternalKeyContextImpl实例的currentKey变量中：

1. 代码读到这里，对我后面的纳闷，您应该能揣测出答案了：state.value()外面会通过StateBackend的keyContext取出方才保留的key，接下来就能像HashMap那样依据key查出该key的状态了，接下来是欢快的印证咱们揣测的过程；
2. 在<font color=”blue”>state.value()</font>代码地位打断点一次看个明确，如下图，果然，state外面有StateBackend的keyContext对象的援用，拜访方才保留的key就不成问题了：

1. 开展state.value()办法如下，简单明了，间接拿keyContext保留的key作为入参去取对应的状态：

1. 再开展下面的get办法，可见最终是从stateMap中获得的，而这个stateMap的具体实现是CopyOnWriteStateMap类型的实例：

1. 代码读到这里，只剩最初一处须要印证了：更新状态的state.update(current)办法，应该也是以StateBackend的keyContext中的key作为本人的key，再将入参的current作为value，更新到stateMap中，来吧，一起印证这个揣测；
2. 开展办法，看到的是stateTable.put办法（后面刚看过stateTable的get办法，稳了）：

1. stateTable.put办法外面和后面的get办法一样，间接拿keyContext保留的key作为本人的key：

1. 最终是调用了stateMap.put办法，将数据保留在CopyOnWriteStateMap实例中：

1. 得益于Flink代码本身标准、清晰的设计和实现，再加上IDEA弱小的debug性能，整个浏览和剖析过程非常顺利，这其中的播种会逐步在今后深刻学习DataStreamAPI的过程中奏效；

最初，依据下面的剖析过程绘制了一幅简陋的流程图，心愿能帮忙您放慢了解：

欢送关注公众号：程序员欣宸

微信搜寻「程序员欣宸」，我是欣宸，期待与您一起畅游Java世界…
https://github.com/zq2599/blog_demos