前些天发现了一个巨牛的人工智能学习网站,通俗易懂,风趣幽默,忍不住给大家分享一下。点击跳转到网站:https://www.captainai.net/dongkelun
前言
前面的 文章 我们介绍了 StreamGraph
的生成,这个实际上只对应 Flink 作业在逻辑上的执行计划图。Flink 会进一步对 StreamGraph
进行转换,得到另一个执行计划图,即 JobGraph
。
JobVertex
在 StreamGraph
中,每一个算子(Operator) 对应了图中的一个节点(StreamNode)。StreamGraph
会被进一步优化,将多个符合条件的节点串联(Chain) 在一起形成一个节点,从而减少数据在不同节点之间流动所产生的序列化、反序列化、网络传输的开销。多个算子被 chain 在一起的形成的节点在 JobGraph
中对应的就是 JobVertex
。
每个 JobVertex
中包含一个或多个 Operators
。 JobVertex
的主要成员变量包括
1 | /** The ID of the vertex. */ |
其输入是 JobEdge
列表, 输出是 IntermediateDataSet
列表。
JobEdge
在 StramGraph
中,StreamNode
之间是通过 StreamEdge
建立连接的。在 JobEdge
中,对应的是 JobEdge
。
和 StreamEdge
中同时保留了源节点和目标节点 (sourceId 和 targetId)不同,在 JobEdge
中只有源节点的信息。由于 JobVertex
中保存了所有输入的 JobEdge
的信息,因而同样可以在两个节点之间建立连接。更确切地说,JobEdge
是和节点的输出结果相关联的,我们看下 JobEdge
主要的成员变量:
1 | /** The vertex connected to this edge. */ |
其中 ResultPartitionType
表示中间结果的类型,说起来有点抽象,我们看下属性就明白了:
1 |
|
这个要结合 Flink 任务运行时的内存管理机制来看,在后面的文章再进行分析。目前在 Stream 模式下使用的类型是 PIPELINED_BOUNDED(true, true, true)
,上面的三个属性都是 true。
StreamConfig
对于每一个 StreamOperator
, 也就是 StreamGraph
中的每一个 StreamGraph
, 在生成 JobGraph
的过程中 StreamingJobGraphGenerator
都会创建一个对应的 StreamConfig
。
StreamConfig
中保存了这个算子(operator) 在运行时需要的所有配置信息,这些信息都是通过 key/value 的形式存储在 Configuration
中的。例如:
1 | //保存StreamOperator信息 |
从 StreamGraph 到 JobGraph
从 StreamGraph
到 JobGraph
的转换入口在 StreamingJobGraphGenerator
中。
首先来看下 StreamingJobGraphGenerator
的成员变量和入口函数:
1 | //id -> JobVertex 的对应关系 |
StreamingJobGraphGenerator#createJobGraph
函数的逻辑也很清晰,首先为所有节点生成一个唯一的hash id,如果节点在多次提交中没有改变(包括并发度、上下游等),那么这个id就不会改变,这主要用于故障恢复。这里我们不能用 StreamNode.id
来代替,因为这是一个从 1 开始的静态计数变量,同样的 Job 可能会得到不一样的 id。然后就是最关键的 chaining 处理,和生成JobVetex
、JobEdge
等。之后就是写入各种配置相关的信息。
我们先来看一下,Flink 是如何确定两个 Operator
是否能够被 chain 到同一个节点的:
1 | //StreamEdge 两端的节点是否能够被 chain 到同一个 JobVertex 中 |
只要一条边两端的节点满足上面的条件,那么这两个节点就可以被串联在同一个 JobVertex
中。接着来就来看最为关键的函数 setChaining 的逻辑:
1 | /** |
上述过程实际上就是通过 DFS 遍历所有的 StreamNode
, 并按照 chainable 的条件不停地将可以串联的 operator 放在同一个的 operator chain 中。每一个 StreamNode
的配置信息都会被序列化到对应的 StreamConfig
中。只有 operator chain 的头部节点会生成对应的 JobVertex
,一个 operator chain 的所有内部节点都会以序列化的形式写入头部节点的 CHAINED_TASK_CONFIG
配置项中。
每一个 operator chain 都会为所有的实际输出边创建对应的 JobEdge
,并和 JobVertex
连接:
1 | private void connect(Integer headOfChain, StreamEdge edge) { |
小结
本文分析了从 StreamGraph
到 JobGraph
之间的转换过程。 JobGraph
的关键在于将多个 StreamNode
优化为一个 JobVertex
, 对应的 StreamEdge
则转化为 JobEdge
, 并且 JobVertex
和 JobEdge
之间通过 IntermediateDataSet
形成一个生产者和消费者的连接关系。
原文地址:https://blog.jrwang.me/2019/flink-source-code-jobgraph/ 作者:jrthe42