Apache Flink Client生成StreamGraph概述节点和边生成StreamGraph小结

上文我们分析提交流程时，<code>RemoteStreamEnvironment</code>类的<code>execute</code>方法的第一步就是生成<code>StreamGraph</code>。

<code>StreamGraph</code>是用于表示流的拓扑结构的数据结构，它包含了生成<code>JobGraph</code>的必要信息。它的类继承关系图如下：

如果你按照<code>StreamGraph</code>的继承链向上追溯，最终会发现它实现了接口<code>FlinkPlan</code>。Flink在这里效仿的是数据库的执行SQL是产生执行计划的机制，<code>FlinkPlan</code>定义在Flink的优化器相关的包中，针对流应用的计划是<code>StreamingPlan</code>。

针对Batch类的应用的计划类是OptimizedPlan。Flink会对Batch类的应用进行优化（这点我们后面会分析），而当前针对Streaming类的应用没有优化措施。

<code>StreamGraph</code>的形象化表示如下图：

上面的图是由“节点”和“边”组成的。节点在Flink中对应的数据结构是<code>StreamNode</code>，而边在Flink中对应的数据结构是<code>StreamEdge</code>。<code>StreamNode</code>和<code>StreamEdge</code>之间存在着组合的依赖关系，依赖关系可见下图：

<code>StreamEdge</code>包含了其连接的源节点<code>sourceVertex</code>和目的节点<code>targetVertex</code>，而<code>StreamNode</code>中包含了与其连接的入边集合<code>inEdges</code>和出边集合<code>outEdges</code>。<code>StreamEdge</code>和<code>StreamNode</code>都有唯一的编号进行标识，但是各自编号的生成规则并不相同。

<code>StreamNode</code>的编号<code>id</code>的生成是通过调用<code>StreamTransformation</code>的静态方法<code>getNewNodeId</code>获得的，其实现是一个静态计数器：

<code>StreamEdge</code>的编号<code>edgeId</code>是字符串类型，其生成的规则为：

它是由多个段连接起来的，语义的文字表述如下：

<code>edgeId</code>除了用来实现StreamEdge的hashCode及equals方法之外并没有其他实际意义。

StreamNode其实是表示operator的数据结构，了解这一点很重要。从Flink开始生成StreamGraph开始，source、sink都是图中的一个节点都是operator，都通过StreamNode这一数据结构来表示，我们常将它们单独拎出来讲是因为它们是流的的输入和输出，但在数据结构层面上它们是一致的。

<code>StreamNode</code>除了存储了输入端和输出端的<code>StreamEdge</code>集合，还封装了<code>operator</code>的其他关键属性，基于这不是我们关注的重点，所以不再赘述。

回过头来我们看<code>JobGraph</code>就不是那么难理解了。它包含了表述整个流拓扑的所有必要信息（比如所有的节点集合、所有的<code>source</code>集合、所有的<code>sink</code>集合、虚拟输出选择节点、虚拟分区节点）。同时还包含了大量操作这些信息的方法。

了解了基础的数据结构之后，我们来分析如何生成<code>JobGraph</code>。定位到<code>getStreamGraph</code>的实现：

它依赖于<code>transformations</code>集合，该集合中存储着一个<code>Streaming</code>程序中所有的转换操作对应的<code>StreamTransformation</code>对象。

每当在<code>DataStream</code>对象上调用<code>transform</code>方法或者调用已经被实现了的一些转换操作（如map、flter等，这些转换操作在内部也调用了<code>transform</code>方法），这些调用都会被加入到<code>transformations</code>集合中。

StreamTransformation表示创建DataStream的操作，其实每个DataStream底层都对应着一个StreamTransformation。DataStream持有执行环境对象的引用，当调用transform方法时，它会调用执行环境对象的addOperator方法，将特定的StreamTransformation对象加入到transformations集合中去，这就是transformations集合中元素的来源。

到目前为止我们提到了多个名词，它们之前拥有着强依赖关系，为了避免混淆，我们以flatMap转换操作为例图示各种对象之间的构建关系：

在源码中，其实Flink自身的命名也并不是那么准确，比如上图中的SingleOutputStreamOperator其实是一种DataStream，但却以Operator结尾，让人匪夷所思。这种情况下，鉴定它们类型的方式可以通过查看它们的继承链来进行识别。

<code>StreamGraph</code>的生成依赖于生成器<code>StreamGraphGenerator</code>，每调用一次静态方法<code>generate</code>才会在内部创建一个<code>StreamGraphGenerator</code>的实例，一个实例对应着一个<code>StreamGraph</code>对象。<code>StreamGraphGenerator</code>调用内部的实例方法<code>generateInternal</code>来遍历<code>transformations</code>集合的每个对象：

在<code>transform</code>方法中，它枚举了Flink中每一种转换类型，并对当前传入的转换类型进行判断，然后将其分发给特定的转换方法进行转换，最终返回当前<code>StreamGraph</code>对象中跟该转换有关的节点编号集合。

你可以将整个过程看作是玩拼图游戏，每遍历完一个转换对象，就离构建完整的<code>StreamGraph</code>更近一步。所有类型各异的转换操作各自持有整个<code>StreamGraph</code>的一部分小图片，根据不同的转换操作类型，它们为<code>StreamGraph</code>提供的“部件”并不完全相同，有的转换只构建节点（如<code>SourceTransformation</code>），有的转换除了构建节点还构建边（如<code>SinkTransformation</code>），有的只构建虚拟节点（如<code>PartitionTransformation</code>、<code>SplitTransformation</code>、<code>SelectTransformation</code>）。

关于虚拟节点，这里需要说明的是并非所有转换操作都具有实际的物理意义（即物理上对应<code>operator</code>）。有些转换操作只具有逻辑概念，例如<code>union</code>，<code>split</code>，<code>select</code>，<code>partition</code>。这些转换操作不会构建真实的<code>StreamNode</code>对象。比如某个流处理应用对应的转换树如下图：

但在运行时，其生成的执行计划，这里也就等同于<code>StreamGraph</code>却是下图这种形式：

从图中可以看到，转换图中对应的一些逻辑操作在产生的执行计划时并不存在，Flink将这些逻辑转换操作转换成了虚拟节点，它们的信息会被绑定到从<code>source</code>到<code>map</code>转换的这条边上。

在给<code>StreamGraph</code>创建并添加一个<code>operator</code>时，需要给该<code>operator</code>指定<code>slotSharingGroup</code>，这时需要调用方法<code>determineSlotSharingGroup</code>来获得SlotSharingGroup的名称：

当用户指定了组名，则直接使用用户指定的名称。如果用户没有指定特定的名称，则需要结合输入节点来做决定：第一种情况如果所有的输入节点都拥有相同的<code>slotSharingGroup</code>名称，那么就使用该组名；否则组名将被命名为<code>default</code>。

Flink当前对于流处理的应用是不作优化的，所以其执行计划就是<code>StreamGraph</code>。Flink提供了一个执行计划的可视化器，它将客户端生成的执行计划以图形的方式展示出来，就像本节开始我们展示的那幅图就是可视化器生成的。那么我们怎么来查看我们自己编写的程序的执行计划呢？其实很简单，我们以Flink的flink-examples-streaming包中的<code>SocketTextStreamWordCount</code>为例，来看一下如何生成执行计划。

我们将<code>SocketTextStreamWordCount</code>最后一行代码注释掉：

然后将其替换成下面这句：

这行语句的作用是打印当前这个程序的执行计划，它将在控制台产生该执行计划的JSON格式表示：

把上面这段JSON复制到Flink的执行计划可视化器，点击下方的<code>Draw</code>按钮，即可生成。

本文我们谈论了<code>StreamGraph</code>的数据结构以及<code>StreamGraphGenerator</code>如何生成<code>StreamGraph</code>。鉴于<code>StreamEdge</code>和<code>StreamNode</code>是组成<code>StreamGraph</code>不可或缺的部分，我们还对这两个数据结构进行了简单的分析。当然，<code>StreamGraph</code>还有一个关键的实例方法：<code>getJobGraph</code>，它用于获取流处理程序的<code>JobGraph</code>（该方法继承自<code>StreamingPlan</code>）。至于什么是<code>JobGraph</code>以及如何获取它，我们将在下文进行讨论。

原文发布时间为：2016-07-23

本文作者：vinoYang