Flink源码阅读：如何生成JobGraph

前文我们介绍了 Flink 的四种执行图，并且通过源码了解了 Flink 的 StreamGraph 是怎么生成的，本文我们就一起来看下 Flink 的另一种执行图——JobGraph 是如何生成的。
StreamGraph 和 JobGraph 的区别

在正式开始之前，我们再来回顾一下 StreamGraph 和 JobGraph 的区别。假设我们的任务是建造一座大楼，StreamGraph 就像是设计蓝图，它描述了每个窗户、每根水管的位置和规格，而 JobGraph 像是给到施工队的施工流程图，它描述了每个任务模块，例如先把地基浇筑好，再铺设管线等。总的来说，JobGraph 更偏向执行层面，它是由 StreamGraph 优化而来。
回到 Flink 本身，我们通过一个表格来了解两个图的区别。
   StreamGraph JobGraph     生成阶段 客户端，执行 execute() 时 客户端，提交前由 StreamGraph 转换生成   抽象层级 高层逻辑图，直接对应 API 优化后的执行图，为调度做准备   核心优化 无 主要是算子链优化   节点 StreamNode JobVertex   边 StreamEdge JobEdge   提交对象 无 提交给 JobManager   包含资源 无 包含执行作业所需的 Jar 包、依赖库和资源文件  JobVertex

JobGraph 中的节点是 JobVertex，在 StreamGraph 转换成 JobGraph 的过程中，会将多个节点串联起来，最终生成 JobVertex。
JobVertex包含以下成员变量：

我们分别看一下这些成员变量及其作用。
1、标识符相关

1. // JobVertex的id，在作业执行过程中的唯一标识。监控、调度和故障恢复都会使用
2. private final JobVertexID id;
4. // operator id列表，按照深度优先顺序存储。operator 的管理、状态分配都会用到
5. private final List<OperatorIDPair> operatorIDs;

复制代码

2、输入输出相关

1. // 定义所有的输入边
2. private final List<JobEdge> inputs = new ArrayList<>();
4. // 定义所有的输出数据集
5. private final Map<IntermediateDataSetID, IntermediateDataSet> results = new LinkedHashMap<>();
7. // 输入分片源，主要用于批处理作业，定义如何将数据分成多个片
8. private InputSplitSource<?> inputSplitSource;

复制代码

3、执行配置相关

1. // 并行度，即运行时拆分子任务数量，默认使用全局配置
2. private int parallelism = ExecutionConfig.PARALLELISM_DEFAULT;
4. // 最大并行度
5. private int maxParallelism = MAX_PARALLELISM_DEFAULT;
7. // 存储运行时实际执行的类，使 Flink 可以灵活处理不同类型的操作符
8. // 流任务可以是"org.apache.flink.streaming.runtime.tasks.StreamTask"
9. // 批任务可以是"org.apache.flink.runtime.operators.BatchTask"
10. private String invokableClassName;
12. // 自定义配置
13. private Configuration configuration;
15. // 是否是动态设置并发度
16. private boolean dynamicParallelism = false;
18. // 是否支持优雅停止
19. private boolean isStoppable = false;

复制代码

4、资源管理相关

1. // JobVertex 最小资源需求
2. private ResourceSpec minResources = ResourceSpec.DEFAULT;
4. // JobVertex 推荐资源需求
5. private ResourceSpec preferredResources = ResourceSpec.DEFAULT;
7. // 用于资源优化，运行不同的 JobVertex 的子任务运行在同一个 slot
8. @Nullable private SlotSharingGroup slotSharingGroup;
10. // 需要严格共址的 JobVertex 组，每个 JobVertex 的第 n 个子任务运行在同一个 TaskManager
11. @Nullable private CoLocationGroupImpl coLocationGroup;

复制代码

5、协调器

1. // 操作符协调器，用于处理全局协调逻辑
2. private final List<SerializedValue<OperatorCoordinator.Provider>> operatorCoordinators =
3.             new ArrayList<>();

复制代码

6、显示和描述信息

1. // JobVertex 的名称
2. private String name;
4. // 操作符名称，比如 'Flat Map' 或 'Join'
5. private String operatorName;
7. // 操作符的描述，比如 'Hash Join' 或 'Sorted Group Reduce'
8. private String operatorDescription;
10. // 提供比 name 更友好的描述信息
11. private String operatorPrettyName;

复制代码

7、状态和行为标志

1. // 是否支持同一个子任务并发多次执行
2. private boolean supportsConcurrentExecutionAttempts = true;
4. // 标记并发度是否被显式设置
5. private boolean parallelismConfigured = false;
7. // 是否有阻塞型输出
8. private boolean anyOutputBlocking = false;

复制代码

8、缓存数据集

1. // 存储该 JobVertex 需要消费的缓存中间数据集的 ID，可提高作业执行效率
2. private final List<IntermediateDataSetID> intermediateDataSetIdsToConsume = new ArrayList<>();

复制代码

JobEdge

在 StreamGraph 中，StreamEdge 是连接 StreamNode 的桥梁。在 JobGraph 中，与之对应的是 JobEdge，不同点在于 JobEdge 中保存的是输入节点和输出结果。
1、连接关系成员

1. // 定义数据流向哪个 JobVertex
2. private final JobVertex target;
4. // 定义这条边的源数据
5. private final IntermediateDataSet source;
7. // 输入类型的编号
8. private final int typeNumber;
10. // 多个输入间的键是否相关，如果为 true，相同键的数据在一个输入被分割时，在其他数据对应的记录也会发送到相同的下游节点
11. private final boolean interInputsKeysCorrelated;
13. // 同一输入内相同的键是否必须发送到同一下游任务
14. private final boolean intraInputKeyCorrelated;

复制代码

2、数据分发模式

1. // 定义数据在并行任务期间的分发模式
2. // 可能值：
3. // ALL_TO_ALL：全连接，每个上游子任务连接所有下游任务
4. // POINTWISE：点对点连接，一对一或一对多的本地连接
5. private final DistributionPattern distributionPattern;

复制代码

3、数据传输策略

1. // 是否为广播连接
2. private final boolean isBroadcast;
4. // 是否为 forward 连接，forward 连接最高效，直接转发，无需序列化网络传输
5. private final boolean isForward;
7. // 数据传输策略名称，用于显示
8. private String shipStrategyName;

复制代码

4、状态重分布映射器

1. // 下游状态重分布映射器，当作业扩容时，决定是否重新分配下游算子的持久化状态
2. private SubtaskStateMapper downstreamSubtaskStateMapper = SubtaskStateMapper.ROUND_ROBIN;
4. // 上游状态重分布映射器，当作业扩容时，决定是否重新分配上游算子的持久化状态
5. private SubtaskStateMapper upstreamSubtaskStateMapper = SubtaskStateMapper.ROUND_ROBIN;

复制代码

5、描述和缓存信息

1. // 预处理操作的名称
2. private String preProcessingOperationName;
4. // 操作符级别缓存的描述
5. private String operatorLevelCachingDescription;

复制代码

StreamGraph 转换成 JobGraph

现在我们再来看一下 StreamGraph 是如何转换成 JobGraph 的。转换逻辑的入口是 StreamGraph.getJobGraph 方法。它只是调用了 StreamingJobGraphGenerator.createJobGraph，核心逻辑在 createJobGraph 方法中。

1. private JobGraph createJobGraph() {
2.     // 预验证，检查 StreamGraph 配置正确性
3.     preValidate(streamGraph, userClassloader);
5.     // 【核心】链化操作符
6.     setChaining();
8.     if (jobGraph.isDynamic()) {
9.         // 支持动态扩缩容场景，为动态图设置并行度
10.         setVertexParallelismsForDynamicGraphIfNecessary();
11.     }
13.     // Note that we set all the non-chainable outputs configuration here because the
14.     // "setVertexParallelismsForDynamicGraphIfNecessary" may affect the parallelism of job
15.     // vertices and partition-reuse
16.     final Map<Integer, Map<StreamEdge, NonChainedOutput>> opIntermediateOutputs =
17.             new HashMap<>();
18.     // 设置不能链化的输出边
19.     setAllOperatorNonChainedOutputsConfigs(opIntermediateOutputs, jobVertexBuildContext);
20.     setAllVertexNonChainedOutputsConfigs(opIntermediateOutputs);
22.     // 设置物理边连接
23.     setPhysicalEdges(jobVertexBuildContext);
25.     // 设置支持并发执行的 JobVertex
26.     markSupportingConcurrentExecutionAttempts(jobVertexBuildContext);
28.     // 验证混合 shuffle 模式只在批处理模式下使用
29.     validateHybridShuffleExecuteInBatchMode(jobVertexBuildContext);
31.     // 设置 Slot 共享和协同定位
32.     setSlotSharingAndCoLocation(jobVertexBuildContext);
34.     // 设置托管内存比例
35.     setManagedMemoryFraction(jobVertexBuildContext);
37.     // 为 JobVertex 名称添加前缀
38.     addVertexIndexPrefixInVertexName(jobVertexBuildContext, new AtomicInteger(0));
40.     // 设置操作符描述信息
41.     setVertexDescription(jobVertexBuildContext);
43.     // Wait for the serialization of operator coordinators and stream config.
44.     // 序列化操作符协调器和流配置
45.     serializeOperatorCoordinatorsAndStreamConfig(serializationExecutor, jobVertexBuildContext);
47.     return jobGraph;
48. }

复制代码

可以看到，在 createJobGraph 方法中，调用了 setChaining 方法，即进行链化操作。这也是 JobGraph 最核心的优化之一。下面我们来看一下具体怎么做链化。

1. private void setChaining() {
2.     // we separate out the sources that run as inputs to another operator (chained inputs)
3.     // from the sources that needs to run as the main (head) operator.
4.     final Map<Integer, OperatorChainInfo> chainEntryPoints =
5.             buildChainedInputsAndGetHeadInputs();
6.     final Collection<OperatorChainInfo> initialEntryPoints =
7.             chainEntryPoints.entrySet().stream()
8.                     .sorted(Comparator.comparing(Map.Entry::getKey))
9.                     .map(Map.Entry::getValue)
10.                     .collect(Collectors.toList());
12.     // iterate over a copy of the values, because this map gets concurrently modified
13.     for (OperatorChainInfo info : initialEntryPoints) {
14.         createChain(
15.                 info.getStartNodeId(),
16.                 1, // operators start at position 1 because 0 is for chained source inputs
17.                 info,
18.                 chainEntryPoints,
19.                 true,
20.                 serializationExecutor,
21.                 jobVertexBuildContext,
22.                 null);
23.     }
24. }

复制代码

setChaining 方法中主要分为两步，第一步是处理 Source 节点，将可以链化的 Source 和不能链化的 Source 节点分开。先来看如何判断一个 Source 是否可被链化。

1. public static boolean isChainableSource(StreamNode streamNode, StreamGraph streamGraph) {
2.     // 最基本的一些判空，输出边数量为1
3.     if (streamNode.getOperatorFactory() == null
4.             || !(streamNode.getOperatorFactory() instanceof SourceOperatorFactory)
5.             || streamNode.getOutEdges().size() != 1) {
6.         return false;
7.     }
8.     final StreamEdge sourceOutEdge = streamNode.getOutEdges().get(0);
9.     final StreamNode target = streamGraph.getStreamNode(sourceOutEdge.getTargetId());
10.     final ChainingStrategy targetChainingStrategy =
11.             Preconditions.checkNotNull(target.getOperatorFactory()).getChainingStrategy();
12.     // 链化策略必须 HEAD_WITH_SOURCES，输出边是可链化的
13.     return targetChainingStrategy == ChainingStrategy.HEAD_WITH_SOURCES
14.             && isChainableInput(sourceOutEdge, streamGraph, false);
15. }
19. private static boolean isChainableInput(
20.         StreamEdge edge, StreamGraph streamGraph, boolean allowChainWithDefaultParallelism) {
21.     StreamNode upStreamVertex = streamGraph.getSourceVertex(edge);
22.     StreamNode downStreamVertex = streamGraph.getTargetVertex(edge);
24.     if (!(streamGraph.isChainingEnabled()
25.             // 上下游节点是否在同一个 slot 共享组
26.             && upStreamVertex.isSameSlotSharingGroup(downStreamVertex)
27.             // 操作符是否可以链化，主要做并行度检查
28.             && areOperatorsChainable(
29.                     upStreamVertex,
30.                     downStreamVertex,
31.                     streamGraph,
32.                     allowChainWithDefaultParallelism)
33.             // 分区器和交换模式是否支持链化
34.             && arePartitionerAndExchangeModeChainable(
35.                     edge.getPartitioner(), edge.getExchangeMode(), streamGraph.isDynamic()))) {
37.         return false;
38.     }
40.     // check that we do not have a union operation, because unions currently only work
41.     // through the network/byte-channel stack.
42.     // we check that by testing that each "type" (which means input position) is used only once
43.     // 检查是否为 Union 操作，Union 操作不能链化
44.     for (StreamEdge inEdge : downStreamVertex.getInEdges()) {
45.         if (inEdge != edge && inEdge.getTypeNumber() == edge.getTypeNumber()) {
46.             return false;
47.         }
48.     }
49.     return true;
50. }

复制代码

Source 的链化条件主要就是这些，我们结合一些例子来看一下。

1. Source(并行度=4) -> Map(并行度=4) -> Filter(并行度=4)
3. Source -> Map 边：
4. 1. isChainingEnabled() = true
5. 2. isSameSlotSharingGroup() = true (都在默认组)
6. 3. areOperatorsChainable() = true (Source可链化，Map是HEAD_WITH_SOURCES)
7. 4. arePartitionerAndExchangeModeChainable() = true (ForwardPartitioner)
8. 5. Union检查通过
9. 结果：可链化
11. Map -> Filter 边：
12. 1. isChainingEnabled() = true
13. 2. isSameSlotSharingGroup() = true
14. 3. areOperatorsChainable() = true (Map和Filter都是ALWAYS)
15. 4. arePartitionerAndExchangeModeChainable() = true (ForwardPartitioner)
16. 5. Union检查通过
17. 结果：可链化
19. 最终：Source -> Map -> Filter 三者链化到一个JobVertex中
22. Source(并行度=2) -> Map(并行度=4)  // 并行度不匹配
24. Source -> Map 边：
25. 1. isChainingEnabled() = true
26. 2. isSameSlotSharingGroup() = true
27. 3. areOperatorsChainable() = false (并行度不匹配)
28. 结果：不可链化，需要网络传输
32. Source1 --\
33.           Union -> Map
34. Source2 --/
36. Source1 -> Union 边：
37. 虽然满足前4个条件，但Union节点有两个输入边，typeNumber相同
38. Union检查失败，不可链化

复制代码

得到了所有入口之后，就可以进行后续节点的链化操作了，它的逻辑在 createChain 方法中。这里主要是一个递归过程，先将节点的输出边分为可链化和不可链化两个 list，之后对可链化的边进行递归调用链化。对不可链化的边，需要创建出新的链。由于篇幅原因，这里只贴一部分核心的代码

1. public static List<StreamEdge> createChain(
2.         final Integer currentNodeId,
3.         final int chainIndex,
4.         final OperatorChainInfo chainInfo,
5.         final Map<Integer, OperatorChainInfo> chainEntryPoints,
6.         final boolean canCreateNewChain,
7.         final Executor serializationExecutor,
8.         final JobVertexBuildContext jobVertexBuildContext,
9.         final @Nullable Consumer<Integer> visitedStreamNodeConsumer) {
11.     ......
13.         // 拆分可链化边和不可链化边
14.         for (StreamEdge outEdge : currentNode.getOutEdges()) {
15.             if (isChainable(outEdge, streamGraph)) {
16.                 chainableOutputs.add(outEdge);
17.             } else {
18.                 nonChainableOutputs.add(outEdge);
19.             }
20.         }
22.         // 处理可链化边
23.         for (StreamEdge chainable : chainableOutputs) {
24.             StreamNode targetNode = streamGraph.getStreamNode(chainable.getTargetId());
25.             Attribute targetNodeAttribute = targetNode.getAttribute();
26.             if (isNoOutputUntilEndOfInput) {
27.                 if (targetNodeAttribute != null) {
28.                     targetNodeAttribute.setNoOutputUntilEndOfInput(true);
29.                 }
30.             }
31.             transitiveOutEdges.addAll(
32.                     createChain(
33.                             chainable.getTargetId(),
34.                             chainIndex + 1,
35.                             chainInfo,
36.                             chainEntryPoints,
37.                             canCreateNewChain,
38.                             serializationExecutor,
39.                             jobVertexBuildContext,
40.                             visitedStreamNodeConsumer));
41.             // Mark upstream nodes in the same chain as outputBlocking
42.             if (targetNodeAttribute != null
43.                     && targetNodeAttribute.isNoOutputUntilEndOfInput()) {
44.                 currentNodeAttribute.setNoOutputUntilEndOfInput(true);
45.             }
46.         }
48.         // 处理不可链化边
49.         for (StreamEdge nonChainable : nonChainableOutputs) {
50.             transitiveOutEdges.add(nonChainable);
51.             // Used to control whether a new chain can be created, this value is true in the
52.             // full graph generation algorithm and false in the progressive generation
53.             // algorithm. In the future, this variable can be a boolean type function to adapt
54.             // to more adaptive scenarios.
55.             if (canCreateNewChain) {
56.                 createChain(
57.                         nonChainable.getTargetId(),
58.                         1, // operators start at position 1 because 0 is for chained source
59.                         // inputs
60.                         chainEntryPoints.computeIfAbsent(
61.                                 nonChainable.getTargetId(),
62.                                 (k) -> chainInfo.newChain(nonChainable.getTargetId())),
63.                         chainEntryPoints,
64.                         canCreateNewChain,
65.                         serializationExecutor,
66.                         jobVertexBuildContext,
67.                         visitedStreamNodeConsumer);
68.             }
69.         }
71.         // 创建 JobVertex
72.         StreamConfig config;
73.         if (currentNodeId.equals(startNodeId)) {
74.             JobVertex jobVertex = jobVertexBuildContext.getJobVertex(startNodeId);
75.             if (jobVertex == null) {
76.                 jobVertex =
77.                         createJobVertex(
78.                                 chainInfo, serializationExecutor, jobVertexBuildContext);
79.             }
80.             config = new StreamConfig(jobVertex.getConfiguration());
81.         } else {
82.             config = new StreamConfig(new Configuration());
83.         }
85.         // 判断是否为起始节点，如果不是，将对应的配置信息存到链化起始节点的 key 中
86.         if (currentNodeId.equals(startNodeId)) {
87.             chainInfo.setTransitiveOutEdges(transitiveOutEdges);
88.             jobVertexBuildContext.addChainInfo(startNodeId, chainInfo);
90.             config.setChainStart();
91.             config.setChainIndex(chainIndex);
92.             config.setOperatorName(streamGraph.getStreamNode(currentNodeId).getOperatorName());
93.             config.setTransitiveChainedTaskConfigs(
94.                     jobVertexBuildContext.getChainedConfigs().get(startNodeId));
95.         } else {
96.             config.setChainIndex(chainIndex);
97.             StreamNode node = streamGraph.getStreamNode(currentNodeId);
98.             config.setOperatorName(node.getOperatorName());
99.             jobVertexBuildContext
100.                     .getOrCreateChainedConfig(startNodeId)
101.                     .put(currentNodeId, config);
102.         }
104.     ......
105. }

复制代码

是否可链化依赖于 isChainable 方法的结果。它主要判断了下游的输入边数量是否为1，然后调用了 isChainableInput，这个方法我们刚刚已经看过了。

1. public static boolean isChainable(StreamEdge edge, StreamGraph streamGraph) {
2.     return isChainable(edge, streamGraph, false);
3. }
5. public static boolean isChainable(
6.         StreamEdge edge, StreamGraph streamGraph, boolean allowChainWithDefaultParallelism) {
7.     StreamNode downStreamVertex = streamGraph.getTargetVertex(edge);
9.     return downStreamVertex.getInEdges().size() == 1
10.             && isChainableInput(edge, streamGraph, allowChainWithDefaultParallelism);
11. }

复制代码

总结

本文我们主要介绍了生成 JobGraph 的相关代码。首先了解了 JobGraph 中的节点和边对应的类，以及它们和 StreamGraph 中的类的映射关系。然后又看了生成 JobGraph 的核心代码，其中重点学习了链化相关的代码。
最后补充一个生成 JobGraph 的调用链路，感兴趣的同学可以看下。

1. clusterClient.submitJob() → MiniCluster.submitJob() → Dispatcher.submitJob() → JobMasterServiceLeadershipRunnerFactory → DefaultJobMasterServiceFactory → JobMaster → DefaultSchedulerFactory.createInstance()→ StreamGraph.getJobGraph()

复制代码

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！