Flink源码阅读：如何生成ExecutionGraph

今天我们一起来了解 Flink 最后一种执行图，ExecutionGraph 的执行过程。
基本概念

在阅读源码之前，我们先来了解一下 ExecutionGraph 中的一些基本概念。

• ExecutionJobVertex: ExecutionJobVertex 是 ExecutionGraph 中的节点，对应的是 JobGraph 中的 JobVertex。
  
• ExecutionVertex: 每个 ExecutionJobVertex 都包含了一组 ExecutionVertex，ExecutionVertex 的数量就是节点对应的并行度。
  
• IntermediateResult: IntermediateResult 表示节点的输出结果，与之对应的是 JobGraph 中的 IntermediateDataSet。
  
• IntermediateResultPartition: IntermediateResultPartition 是每个 ExecutionVertex 的输出。
  
• EdgeManager: EdgeManager 主要负责存储 ExecutionGraph 中所有之间的连接，包括其并行度。
  
• Execution: Execution 可以认为是一次实际的运行尝试。每次执行时，Flink 都会将ExecutionVertex 封装成一个 Execution，并通过一个 ExecutionAttemptID 来做唯一标识。
  

ExecutionGraph 生成过程

了解了这些基本概念之后，我们一起来看一下 ExecutionGraph 的具体生成过程。生成 ExecutionGraph 的代码入口是 DefaultExecutionGraphBuilder.build 方法。
首先是获取一些基本信息，包括 jobInformation、jobStatusChangedListeners 等。
接下来就是创建一个 DefaultExecutionGraph 和生成执行计划。

1. // create a new execution graph, if none exists so far
2. final DefaultExecutionGraph executionGraph =
3.         new DefaultExecutionGraph(
4.                 jobInformation,
5.                 futureExecutor,
6.                 ioExecutor,
7.                 rpcTimeout,
8.                 executionHistorySizeLimit,
9.                 classLoader,
10.                 blobWriter,
11.                 partitionGroupReleaseStrategyFactory,
12.                 shuffleMaster,
13.                 partitionTracker,
14.                 executionDeploymentListener,
15.                 executionStateUpdateListener,
16.                 initializationTimestamp,
17.                 vertexAttemptNumberStore,
18.                 vertexParallelismStore,
19.                 isDynamicGraph,
20.                 executionJobVertexFactory,
21.                 jobGraph.getJobStatusHooks(),
22.                 markPartitionFinishedStrategy,
23.                 taskDeploymentDescriptorFactory,
24.                 jobStatusChangedListeners,
25.                 executionPlanSchedulingContext);
28. try {
29.     executionGraph.setPlan(JsonPlanGenerator.generatePlan(jobGraph));
30. } catch (Throwable t) {
31.     log.warn("Cannot create plan for job", t);
32.     // give the graph an empty plan
33.     executionGraph.setPlan(new JobPlanInfo.Plan("", "", "", new ArrayList<>()));
34. }

复制代码

下面就是两个比较核心的方法 getVerticesSortedTopologicallyFromSources 和 attachJobGraph。

1. // topologically sort the job vertices and attach the graph to the existing one
2. List<JobVertex> sortedTopology = jobGraph.getVerticesSortedTopologicallyFromSources();
4. executionGraph.attachJobGraph(sortedTopology, jobManagerJobMetricGroup);

复制代码

这两个方法是先将 JobVertex 进行排序，然后构建 ExecutionGraph 的拓扑图。
getVerticesSortedTopologicallyFromSources

1. public List<JobVertex> getVerticesSortedTopologicallyFromSources()
2.         throws InvalidProgramException {
3.     // early out on empty lists
4.     if (this.taskVertices.isEmpty()) {
5.         return Collections.emptyList();
6.     }
8.     List<JobVertex> sorted = new ArrayList<JobVertex>(this.taskVertices.size());
9.     Set<JobVertex> remaining = new LinkedHashSet<JobVertex>(this.taskVertices.values());
11.     // start by finding the vertices with no input edges
12.     // and the ones with disconnected inputs (that refer to some standalone data set)
13.     {
14.         Iterator<JobVertex> iter = remaining.iterator();
15.         while (iter.hasNext()) {
16.             JobVertex vertex = iter.next();
18.             if (vertex.isInputVertex()) {
19.                 sorted.add(vertex);
20.                 iter.remove();
21.             }
22.         }
23.     }
25.     int startNodePos = 0;
27.     // traverse from the nodes that were added until we found all elements
28.     while (!remaining.isEmpty()) {
30.         // first check if we have more candidates to start traversing from. if not, then the
31.         // graph is cyclic, which is not permitted
32.         if (startNodePos >= sorted.size()) {
33.             throw new InvalidProgramException("The job graph is cyclic.");
34.         }
36.         JobVertex current = sorted.get(startNodePos++);
37.         addNodesThatHaveNoNewPredecessors(current, sorted, remaining);
38.     }
40.     return sorted;
41. }

复制代码

这段代码是将所有的节点进行排序，先将所有的 Source 节点筛选出来，然后再将剩余节点假如列表。这样就能构建出最终的拓扑图。
attachJobGraph

1. @Override
2. public void attachJobGraph(
3.         List<JobVertex> verticesToAttach, JobManagerJobMetricGroup jobManagerJobMetricGroup)
4.         throws JobException {
6.     assertRunningInJobMasterMainThread();
8.     LOG.debug(
9.             "Attaching {} topologically sorted vertices to existing job graph with {} "
10.                     + "vertices and {} intermediate results.",
11.             verticesToAttach.size(),
12.             tasks.size(),
13.             intermediateResults.size());
15.     attachJobVertices(verticesToAttach, jobManagerJobMetricGroup);
16.     if (!isDynamic) {
17.         initializeJobVertices(verticesToAttach);
18.     }
20.     // the topology assigning should happen before notifying new vertices to failoverStrategy
21.     executionTopology = DefaultExecutionTopology.fromExecutionGraph(this);
23.     partitionGroupReleaseStrategy =
24.             partitionGroupReleaseStrategyFactory.createInstance(getSchedulingTopology());
25. }

复制代码

attachJobGraph 方法主要包含两步逻辑，第一步是调用 attachJobVertices 方法创建 ExecutionJobVertex 实例，第二步是调用 fromExecutionGraph 创建一些其他的核心对象。
attachJobVertices
attachJobVertices 方法中就是遍历所有的 JobVertex，然后利用 JobVertex 生成 ExecutionJobVertex。

1. /** Attach job vertices without initializing them. */
2. private void attachJobVertices(
3.         List<JobVertex> topologicallySorted, JobManagerJobMetricGroup jobManagerJobMetricGroup)
4.         throws JobException {
5.     for (JobVertex jobVertex : topologicallySorted) {
7.         if (jobVertex.isInputVertex() && !jobVertex.isStoppable()) {
8.             this.isStoppable = false;
9.         }
11.         VertexParallelismInformation parallelismInfo =
12.                 parallelismStore.getParallelismInfo(jobVertex.getID());
14.         // create the execution job vertex and attach it to the graph
15.         ExecutionJobVertex ejv =
16.                 executionJobVertexFactory.createExecutionJobVertex(
17.                         this,
18.                         jobVertex,
19.                         parallelismInfo,
20.                         coordinatorStore,
21.                         jobManagerJobMetricGroup);
23.         ExecutionJobVertex previousTask = this.tasks.putIfAbsent(jobVertex.getID(), ejv);
24.         if (previousTask != null) {
25.             throw new JobException(
26.                     String.format(
27.                             "Encountered two job vertices with ID %s : previous=[%s] / new=[%s]",
28.                             jobVertex.getID(), ejv, previousTask));
29.         }
31.         this.verticesInCreationOrder.add(ejv);
32.         this.numJobVerticesTotal++;
33.     }
34. }

复制代码

initializeJobVertices
在 DefaultExecutionGraph.initializeJobVertices 中是遍历了刚刚排好序的 JobVertex，获取了 ExecutionJobVertex 之后调用了 ExecutionGraph.initializeJobVertex 方法。
我们直接来看 ExecutionGraph.initializeJobVertex 的逻辑。

1. default void initializeJobVertex(ExecutionJobVertex ejv, long createTimestamp)
2.         throws JobException {
3.     initializeJobVertex(
4.             ejv,
5.             createTimestamp,
6.             VertexInputInfoComputationUtils.computeVertexInputInfos(
7.                     ejv, getAllIntermediateResults()::get));
8. }

复制代码

这里先是调用了 VertexInputInfoComputationUtils.computeVertexInputInfos 方法，生成了 Map jobVertexInputInfos。它表示的是每个 ExecutionVertex 消费上游 IntermediateResultPartition 的范围。
这里有两种模式，分别是 POINTWISE （点对点）和 ALL_TO_ALL（全对全）
在 POINTWISE 模式中，会按照尽量均匀分布的方式处理。

• 例如上游并发度是4，下游并发度是2时，那么前两个 IntermediateResultPartition 就会被第一个 ExecutionVertex 消费，后两个 IntermediateResultPartition 就会被第二个 ExecutionVertex 消费。
  
• 如果上游并发度是2，下游是3时，那么下游前两个 IntermediateResultPartition 会被第一个 ExecutionVertex 消费，第三个 IntermediateResultPartition 则会被第二个 ExecutionVertex 消费。
  

1. public static JobVertexInputInfo computeVertexInputInfoForPointwise(
2.         int sourceCount,
3.         int targetCount,
4.         Function<Integer, Integer> numOfSubpartitionsRetriever,
5.         boolean isDynamicGraph) {
7.     final List<ExecutionVertexInputInfo> executionVertexInputInfos = new ArrayList<>();
9.     if (sourceCount >= targetCount) {
10.         for (int index = 0; index < targetCount; index++) {
12.             int start = index * sourceCount / targetCount;
13.             int end = (index + 1) * sourceCount / targetCount;
15.             IndexRange partitionRange = new IndexRange(start, end - 1);
16.             IndexRange subpartitionRange =
17.                     computeConsumedSubpartitionRange(
18.                             index,
19.                             1,
20.                             () -> numOfSubpartitionsRetriever.apply(start),
21.                             isDynamicGraph,
22.                             false,
23.                             false);
24.             executionVertexInputInfos.add(
25.                     new ExecutionVertexInputInfo(index, partitionRange, subpartitionRange));
26.         }
27.     } else {
28.         for (int partitionNum = 0; partitionNum < sourceCount; partitionNum++) {
30.             int start = (partitionNum * targetCount + sourceCount - 1) / sourceCount;
31.             int end = ((partitionNum + 1) * targetCount + sourceCount - 1) / sourceCount;
32.             int numConsumers = end - start;
34.             IndexRange partitionRange = new IndexRange(partitionNum, partitionNum);
35.             // Variable used in lambda expression should be final or effectively final
36.             final int finalPartitionNum = partitionNum;
37.             for (int i = start; i < end; i++) {
38.                 IndexRange subpartitionRange =
39.                         computeConsumedSubpartitionRange(
40.                                 i,
41.                                 numConsumers,
42.                                 () -> numOfSubpartitionsRetriever.apply(finalPartitionNum),
43.                                 isDynamicGraph,
44.                                 false,
45.                                 false);
46.                 executionVertexInputInfos.add(
47.                         new ExecutionVertexInputInfo(i, partitionRange, subpartitionRange));
48.             }
49.         }
50.     }
51.     return new JobVertexInputInfo(executionVertexInputInfos);
52. }

复制代码

在 ALL_TO_ALL 模式中，每个下游都会消费所有上游的数据。

1. public static JobVertexInputInfo computeVertexInputInfoForAllToAll(
2.         int sourceCount,
3.         int targetCount,
4.         Function<Integer, Integer> numOfSubpartitionsRetriever,
5.         boolean isDynamicGraph,
6.         boolean isBroadcast,
7.         boolean isSingleSubpartitionContainsAllData) {
8.     final List<ExecutionVertexInputInfo> executionVertexInputInfos = new ArrayList<>();
9.     IndexRange partitionRange = new IndexRange(0, sourceCount - 1);
10.     for (int i = 0; i < targetCount; ++i) {
11.         IndexRange subpartitionRange =
12.                 computeConsumedSubpartitionRange(
13.                         i,
14.                         targetCount,
15.                         () -> numOfSubpartitionsRetriever.apply(0),
16.                         isDynamicGraph,
17.                         isBroadcast,
18.                         isSingleSubpartitionContainsAllData);
19.         executionVertexInputInfos.add(
20.                 new ExecutionVertexInputInfo(i, partitionRange, subpartitionRange));
21.     }
22.     return new JobVertexInputInfo(executionVertexInputInfos);
23. }

复制代码

生成好了 jobVertexInputInfos 之后，我们再回到 DefaultExecutionGraph.initializeJobVertex 方法中。

1. @Override
2. public void initializeJobVertex(
3.         ExecutionJobVertex ejv,
4.         long createTimestamp,
5.         Map<IntermediateDataSetID, JobVertexInputInfo> jobVertexInputInfos)
6.         throws JobException {
8.     checkNotNull(ejv);
9.     checkNotNull(jobVertexInputInfos);
11.     jobVertexInputInfos.forEach(
12.             (resultId, info) ->
13.                     this.vertexInputInfoStore.put(ejv.getJobVertexId(), resultId, info));
15.     ejv.initialize(
16.             executionHistorySizeLimit,
17.             rpcTimeout,
18.             createTimestamp,
19.             this.initialAttemptCounts.getAttemptCounts(ejv.getJobVertexId()),
20.             executionPlanSchedulingContext);
22.     ejv.connectToPredecessors(this.intermediateResults);
24.     for (IntermediateResult res : ejv.getProducedDataSets()) {
25.         IntermediateResult previousDataSet =
26.                 this.intermediateResults.putIfAbsent(res.getId(), res);
27.         if (previousDataSet != null) {
28.             throw new JobException(
29.                     String.format(
30.                             "Encountered two intermediate data set with ID %s : previous=[%s] / new=[%s]",
31.                             res.getId(), res, previousDataSet));
32.         }
33.     }
35.     registerExecutionVerticesAndResultPartitionsFor(ejv);
37.     // enrich network memory.
38.     SlotSharingGroup slotSharingGroup = ejv.getSlotSharingGroup();
39.     if (areJobVerticesAllInitialized(slotSharingGroup)) {
40.         SsgNetworkMemoryCalculationUtils.enrichNetworkMemory(
41.                 slotSharingGroup, this::getJobVertex, shuffleMaster);
42.     }
43. }

复制代码

首先来看 ExecutionJobVertex.initialize 方法。这个方法主要是生成 IntermediateResult 和 ExecutionVertex。

1. protected void initialize(
2.         int executionHistorySizeLimit,
3.         Duration timeout,
4.         long createTimestamp,
5.         SubtaskAttemptNumberStore initialAttemptCounts,
6.         ExecutionPlanSchedulingContext executionPlanSchedulingContext)
7.         throws JobException {
9.     checkState(parallelismInfo.getParallelism() > 0);
10.     checkState(!isInitialized());
12.     this.taskVertices = new ExecutionVertex[parallelismInfo.getParallelism()];
14.     this.inputs = new ArrayList<>(jobVertex.getInputs().size());
16.     // create the intermediate results
17.     this.producedDataSets =
18.             new IntermediateResult[jobVertex.getNumberOfProducedIntermediateDataSets()];
20.     for (int i = 0; i < jobVertex.getProducedDataSets().size(); i++) {
21.         final IntermediateDataSet result = jobVertex.getProducedDataSets().get(i);
23.         this.producedDataSets[i] =
24.                 new IntermediateResult(
25.                         result,
26.                         this,
27.                         this.parallelismInfo.getParallelism(),
28.                         result.getResultType(),
29.                         executionPlanSchedulingContext);
30.     }
32.     // create all task vertices
33.     for (int i = 0; i < this.parallelismInfo.getParallelism(); i++) {
34.         ExecutionVertex vertex =
35.                 createExecutionVertex(
36.                         this,
37.                         i,
38.                         producedDataSets,
39.                         timeout,
40.                         createTimestamp,
41.                         executionHistorySizeLimit,
42.                         initialAttemptCounts.getAttemptCount(i));
44.         this.taskVertices[i] = vertex;
45.     }
47.     // sanity check for the double referencing between intermediate result partitions and
48.     // execution vertices
49.     for (IntermediateResult ir : this.producedDataSets) {
50.         if (ir.getNumberOfAssignedPartitions() != this.parallelismInfo.getParallelism()) {
51.             throw new RuntimeException(
52.                     "The intermediate result's partitions were not correctly assigned.");
53.         }
54.     }
56.     // set up the input splits, if the vertex has any
57.     try {
58.         @SuppressWarnings("unchecked")
59.         InputSplitSource<InputSplit> splitSource =
60.                 (InputSplitSource<InputSplit>) jobVertex.getInputSplitSource();
62.         if (splitSource != null) {
63.             Thread currentThread = Thread.currentThread();
64.             ClassLoader oldContextClassLoader = currentThread.getContextClassLoader();
65.             currentThread.setContextClassLoader(graph.getUserClassLoader());
66.             try {
67.                 inputSplits =
68.                         splitSource.createInputSplits(this.parallelismInfo.getParallelism());
70.                 if (inputSplits != null) {
71.                     splitAssigner = splitSource.getInputSplitAssigner(inputSplits);
72.                 }
73.             } finally {
74.                 currentThread.setContextClassLoader(oldContextClassLoader);
75.             }
76.         } else {
77.             inputSplits = null;
78.         }
79.     } catch (Throwable t) {
80.         throw new JobException(
81.                 "Creating the input splits caused an error: " + t.getMessage(), t);
82.     }
83. }

复制代码

在创建 ExecutionVertex 时，会创建 IntermediateResultPartition 和 Execution，创建 Execution 时，会设置 attemptNumber，这个值默认是0，如果 ExecutionVertex 是重新调度的，那么 attemptNumber 会自增加1。
ExecutionJobVertex.connectToPredecessors 方法主要是生成 ExecutionVertex 与 IntermediateResultPartition 的关联关系。这里设置关联关系也分成了点对点和全对全两种模式处理，点对点模式需要计算 ExecutionVertex 对应的 IntermediateResultPartition index 的范围。两种模式最终都调用了 connectInternal 方法。

1. /** Connect all execution vertices to all partitions. */
2. private static void connectInternal(
3.         List<ExecutionVertex> taskVertices,
4.         List<IntermediateResultPartition> partitions,
5.         ResultPartitionType resultPartitionType,
6.         EdgeManager edgeManager) {
7.     checkState(!taskVertices.isEmpty());
8.     checkState(!partitions.isEmpty());
10.     ConsumedPartitionGroup consumedPartitionGroup =
11.             createAndRegisterConsumedPartitionGroupToEdgeManager(
12.                     taskVertices.size(), partitions, resultPartitionType, edgeManager);
13.     for (ExecutionVertex ev : taskVertices) {
14.         ev.addConsumedPartitionGroup(consumedPartitionGroup);
15.     }
17.     List<ExecutionVertexID> consumerVertices =
18.             taskVertices.stream().map(ExecutionVertex::getID).collect(Collectors.toList());
19.     ConsumerVertexGroup consumerVertexGroup =
20.             ConsumerVertexGroup.fromMultipleVertices(consumerVertices, resultPartitionType);
21.     for (IntermediateResultPartition partition : partitions) {
22.         partition.addConsumers(consumerVertexGroup);
23.     }
25.     consumedPartitionGroup.setConsumerVertexGroup(consumerVertexGroup);
26.     consumerVertexGroup.setConsumedPartitionGroup(consumedPartitionGroup);
27. }

复制代码

这个方法中 ev.addConsumedPartitionGroup(consumedPartitionGroup); 负责将 ExecutionVertex 到 IntermediateResultPartition 的关联关系保存在 EdgeManager.vertexConsumedPartitions 中。
而 partition.addConsumers(consumerVertexGroup); 则负责将 IntermediateResultPartition 到 ExecutionVertex 的关系保存在 EdgeManager.partitionConsumers 中。
总结

通过本文，我们了解了 Flink 是如何将 JobGraph 转换成 ExecutionGraph 的。其中涉及到的一些核心概念名称比较类似，建议认真学习和理解透彻之后再研究其生成方法和对应关系，也可以借助前文中 ExecutionGraph 示意图辅助学习。

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