Flink：两个递归彻底搞懂operator chain

operator chain是指将满足一定条件的operator 链在一起，放在同一个task里面执行，是Flink任务优化的一种方式，在同一个task里面的operator的数据传输变成函数调用关系，这种方式减少数据传输过程。常见的chain例如：source->map->filter，这样的任务链可以chain在一起，那么其内部是如何决定是否能够chain在一起与chain一起之后如何执行就是本篇文章将要剖析的重点。

第一个递归：JobGraph生成

Flink中划分了四种图：StreamGraph、JobGraph、ExecutionGraph、物理执行图，前两种StreamGraph、JobGraph是在客户端生成，ExecutionGraph在jobMaster中生成，最后一种物理执行图是一种虚拟的图，不存在的数据结构，运行在每一个TaskExecutor中。我们在Flink Web UI中看到的就是JobGraph，如下：

JobGraph相对于StreamGraph，可以理解为优化过后的StreamGraph，将能够chain一起的operator chain在一起，上图将source与filter两个operator chain在一起了，这个步骤在生成JobGraph过程中完成。其具体实现在StreamingJobGraphGenerator中：

重点就在setChaining方法中，在里面调用createChain方法，构造JobVertix的同时完成operator chain的操作，createChain方法：

整个过程就是一个递归的过程，createChain 过程就是不断寻找一个chain的出边，如果邻接的两个operator(source与filter)能够chain在一起，那么就以下一个能够chain一起的operator(filter)为起点，继续寻找，直到找到不能够chain一起的operator(process1)，但是此时并没有立刻返回，而是以当前不能chain再一起的operator(process1)为起点继续往下寻找，直到终点(sink）开始一层一层返回，实际上其构造过程是一个反向过程：sink->process2->process1->(source&filter) 这样的一个过程完成operator chain并且构造JobVertix (可通过debug方式查看其详细过程)。

如何判断两个相邻的operator(StreamNode)能够chain在一起？通过isChainable方法判断：

第二个递归：OperatorChain生成

当Execution在deploy的过程中，也就是Task在TaskExecutor启动过程中， 会生成一个OperatorChain对象，在该OperatorChain对象中包含了所有的能够chain在一起的operator(source&filter)，其内部会生成一个名为chainEntryPoint的WatermarkGaugeExposingOutput对象，一个将数据输出的对象，其输出有两种形式：

1.函数调用，将数据推送给chain在一起的下一个operator节点(filter中) 2.输出到下一个没有被chain的operator(process1)

那么chainEntryPoint是如何生成的？

OperatorChain的初始化是在StreamTask中被调用的：

在其构造函数中调用：

createOutputCollector获取当前节点的out，

createChainedOperator方法：

可以看出整个构造chainEntryPoint的过程也是一个递归的过程，会不断寻找被chain在一起的下一个operator的out，直到下游没有可chain的位置，返回网络out作为最后一个operator的out，然后通过使用当前operator构造前一个operator的out，同样是一个反向构造out的过程(filterOut->sourceOut)。简化一下逻辑代码：

最终得到的chainEntryPoint就是headOperator的out，这样在其内部不断的通过out调用operator的方式实现了chain的函数调用链关系。

总结

透过以上分析，operator chain就是将满足一定条件的的operator通过函数调用方式传递数据，避免了数据传输的中间过程。