Flink源码剖析之：如何根据JobGraph生成ExecutionGraph

Flink源码剖析之：如何根据JobGraph生成ExecutionGraph

在上一篇Flink源码剖析中，我们介绍了Flink如何根据StreamGraph生成JobGraph的流程，并着重分析了其算子链的归并过程和JobGraph的构造流程。
对于StreamGraph和JobGraph的生成来说，其都是在客户端生成的，本文将会报告JobGraph到ExecutionGraph的生成过程，而这一过程会在Flink JobManager的服务端来完成。当JobGraph从客户端提交到JobManager后，JobManager会根据JobGraph生成对应的ExecutionGraph，而ExecutionGraph就是Flink作业调度时使用的焦点数据结构。 本篇将会详细介绍JobGraph转换为ExecutionGraph的流程。
主体流程梳理

Flink在将JobGraph转换成ExecutionGraph后，便可以开始实行真正的任务。这一转换流程主要在Flink源码中的DefaultExecutionGraphBuilder类中的buildGraph方法中实现的。在转换过程中，涉及到了一些新的根本概念，先来简朴介绍一下这些概念，对于明白ExecutionGraph有较大的资助：

• ExecutionJobVertex: 在ExecutionGraph中表示实行顶点，与JobGraph中的JobVertex逐一对应。实际上，每个ExecutionJobVertex也是依赖JobVertex来创建的。
  
• ExecutionVertex: 在ExecutionJobVertex类中创建，每个并发度都对应了一个ExecutionVertex对象，每个ExecutionVertex都代表JobVertex在某个特定并行子任务中的实行。在实际实行时，每个ExecutionVertex实际上就是一个Task，是ExecutionJobVertex并行实行的一个子任务。
  
• Execution: Execution表示ExecutionVertex的一次实行。由于ExecutionVertex可以被实行多次（用于恢复、重新计算、重新分配），这个类用于跟踪该ExecutionVertex的单个实行状态和资源。
  
• IntermediateResult: 在JobGraph中用IntermediateDataSet表示上游JobVertex的输出数据流，而在ExecutionGraph中，则用IntermediateResult来表示ExecutionJobVertex的输出数据流。
  
• IntermediateResultPartition:这是IntermediateResult的一部分或一个分片。由于有多个并行任务（ExecutionVertex）实行雷同的操作，每个任务都会产生一部分IntermediateResult。这些结果在物理存储和计算过程中，大概会被进一步划分成多个分区，每个分区对应一个 IntermediateResultPartition对象。
  

从上面的根本概念也可以看出，在ExecutionGraph中：

• 相比StreamGraph和JobGraph，ExecutionGraph是实际根据任务并行度来生成拓扑结构的，在ExecutionGraph中，每个并行子任务都对应一个ExecutionVertex顶点和IntermediateResultPartition输出数据流分区。
  
• 在ExecutionGraph中，上下游节点之间的连接是通过ExecutionVertex -> IntermediateResultPartition -> ExecutionVertex 对象来完成的。
  

整体的实行流程图如下所示：

入口方法：DefaultExecutionGraphBuilder.buildGraph

ExecutionGraph的生成是在DefaultExecutionGraphBuilder类的buildGraph方法中实现的：

1. public class DefaultExecutionGraphBuilder {
3.     public static DefaultExecutionGraph buildGraph(
4.             JobGraph jobGraph,
5.             Configuration jobManagerConfig,
6.             ScheduledExecutorService futureExecutor,
7.             Executor ioExecutor,
8.             ClassLoader classLoader,
9.             CompletedCheckpointStore completedCheckpointStore,
10.             CheckpointsCleaner checkpointsCleaner,
11.             CheckpointIDCounter checkpointIdCounter,
12.             Time rpcTimeout,
13.             MetricGroup metrics,
14.             BlobWriter blobWriter,
15.             Logger log,
16.             ShuffleMaster<?> shuffleMaster,
17.             JobMasterPartitionTracker partitionTracker,
18.             TaskDeploymentDescriptorFactory.PartitionLocationConstraint partitionLocationConstraint,
19.             ExecutionDeploymentListener executionDeploymentListener,
20.             ExecutionStateUpdateListener executionStateUpdateListener,
21.             long initializationTimestamp,
22.             VertexAttemptNumberStore vertexAttemptNumberStore,
23.             VertexParallelismStore vertexParallelismStore)
24.             throws JobExecutionException, JobException {
26.         checkNotNull(jobGraph, "job graph cannot be null");
28.         final String jobName = jobGraph.getName();
29.         final JobID jobId = jobGraph.getJobID();
31.                 // 创建JobInformation
32.         final JobInformation jobInformation =
33.                 new JobInformation(
34.                         jobId,
35.                         jobName,
36.                         jobGraph.getSerializedExecutionConfig(),
37.                         jobGraph.getJobConfiguration(),
38.                         jobGraph.getUserJarBlobKeys(),
39.                         jobGraph.getClasspaths());
40.                
41.                 // Execution 保留的最大历史数
42.         final int maxPriorAttemptsHistoryLength =
43.                 jobManagerConfig.getInteger(JobManagerOptions.MAX_ATTEMPTS_HISTORY_SIZE);
45.                 // IntermediateResultPartitions的释放策略
46.         final PartitionGroupReleaseStrategy.Factory partitionGroupReleaseStrategyFactory =
47.                 PartitionGroupReleaseStrategyFactoryLoader.loadPartitionGroupReleaseStrategyFactory(
48.                         jobManagerConfig);
50.         // create a new execution graph, if none exists so far
51.         final DefaultExecutionGraph executionGraph;
52.         try {
53.                         // 创建默认的ExecutionGraph执行图对象，最后会返回该创建好的执行图对象
54.             executionGraph =
55.                     new DefaultExecutionGraph(
56.                             jobInformation,
57.                             futureExecutor,
58.                             ioExecutor,
59.                             rpcTimeout,
60.                             maxPriorAttemptsHistoryLength,
61.                             classLoader,
62.                             blobWriter,
63.                             partitionGroupReleaseStrategyFactory,
64.                             shuffleMaster,
65.                             partitionTracker,
66.                             partitionLocationConstraint,
67.                             executionDeploymentListener,
68.                             executionStateUpdateListener,
69.                             initializationTimestamp,
70.                             vertexAttemptNumberStore,
71.                             vertexParallelismStore);
72.         } catch (IOException e) {
73.             throw new JobException("Could not create the ExecutionGraph.", e);
74.         }
76.         // set the basic properties
78.         try {
79.             executionGraph.setJsonPlan(JsonPlanGenerator.generatePlan(jobGraph));
80.         } catch (Throwable t) {
81.             log.warn("Cannot create JSON plan for job", t);
82.             // give the graph an empty plan
83.             executionGraph.setJsonPlan("{}");
84.         }
86.         // initialize the vertices that have a master initialization hook
87.         // file output formats create directories here, input formats create splits
89.         final long initMasterStart = System.nanoTime();
90.         log.info("Running initialization on master for job {} ({}).", jobName, jobId);
92.         for (JobVertex vertex : jobGraph.getVertices()) {
93.             String executableClass = vertex.getInvokableClassName();
94.             if (executableClass == null || executableClass.isEmpty()) {
95.                 throw new JobSubmissionException(
96.                         jobId,
97.                         "The vertex "
98.                                 + vertex.getID()
99.                                 + " ("
100.                                 + vertex.getName()
101.                                 + ") has no invokable class.");
102.             }
104.             try {
105.                 vertex.initializeOnMaster(classLoader);
106.             } catch (Throwable t) {
107.                 throw new JobExecutionException(
108.                         jobId,
109.                         "Cannot initialize task '" + vertex.getName() + "': " + t.getMessage(),
110.                         t);
111.             }
112.         }
114.         log.info(
115.                 "Successfully ran initialization on master in {} ms.",
116.                 (System.nanoTime() - initMasterStart) / 1_000_000);
118.         // topologically sort the job vertices and attach the graph to the existing one
119.                 // 这里会先做一个排序，source源节点会放在最前面，接着开始遍历
120.                 // 必须保证当前添加到集合的节点的前置节点都已经添加进去了
121.         List<JobVertex> sortedTopology = jobGraph.getVerticesSortedTopologicallyFromSources();
122.         if (log.isDebugEnabled()) {
123.             log.debug(
124.                     "Adding {} vertices from job graph {} ({}).",
125.                     sortedTopology.size(),
126.                     jobName,
127.                     jobId);
128.         }
130.                 // 构建执行图的重点方法。生成具体的ExecutionGraph
131.         executionGraph.attachJobGraph(sortedTopology);
133.         if (log.isDebugEnabled()) {
134.             log.debug(
135.                     "Successfully created execution graph from job graph {} ({}).", jobName, jobId);
136.         }
138.         // configure the state checkpointing
139.                 // checkpoint的相关配置
140.         if (isCheckpointingEnabled(jobGraph)) {
141.             JobCheckpointingSettings snapshotSettings = jobGraph.getCheckpointingSettings();
143.             // Maximum number of remembered checkpoints
144.             int historySize = jobManagerConfig.getInteger(WebOptions.CHECKPOINTS_HISTORY_SIZE);
146.             CheckpointStatsTracker checkpointStatsTracker =
147.                     new CheckpointStatsTracker(
148.                             historySize,
149.                             snapshotSettings.getCheckpointCoordinatorConfiguration(),
150.                             metrics);
152.             // load the state backend from the application settings
153.             final StateBackend applicationConfiguredBackend;
154.             final SerializedValue<StateBackend> serializedAppConfigured =
155.                     snapshotSettings.getDefaultStateBackend();
157.             if (serializedAppConfigured == null) {
158.                 applicationConfiguredBackend = null;
159.             } else {
160.                 try {
161.                     applicationConfiguredBackend =
162.                             serializedAppConfigured.deserializeValue(classLoader);
163.                 } catch (IOException | ClassNotFoundException e) {
164.                     throw new JobExecutionException(
165.                             jobId, "Could not deserialize application-defined state backend.", e);
166.                 }
167.             }
169.                         // StateBackend配置
170.             final StateBackend rootBackend;
171.             try {
172.                 rootBackend =
173.                         StateBackendLoader.fromApplicationOrConfigOrDefault(
174.                                 applicationConfiguredBackend,
175.                                 snapshotSettings.isChangelogStateBackendEnabled(),
176.                                 jobManagerConfig,
177.                                 classLoader,
178.                                 log);
179.             } catch (IllegalConfigurationException | IOException | DynamicCodeLoadingException e) {
180.                 throw new JobExecutionException(
181.                         jobId, "Could not instantiate configured state backend", e);
182.             }
184.             // load the checkpoint storage from the application settings
185.             final CheckpointStorage applicationConfiguredStorage;
186.             final SerializedValue<CheckpointStorage> serializedAppConfiguredStorage =
187.                     snapshotSettings.getDefaultCheckpointStorage();
189.             if (serializedAppConfiguredStorage == null) {
190.                 applicationConfiguredStorage = null;
191.             } else {
192.                 try {
193.                     applicationConfiguredStorage =
194.                             serializedAppConfiguredStorage.deserializeValue(classLoader);
195.                 } catch (IOException | ClassNotFoundException e) {
196.                     throw new JobExecutionException(
197.                             jobId,
198.                             "Could not deserialize application-defined checkpoint storage.",
199.                             e);
200.                 }
201.             }
203.             final CheckpointStorage rootStorage;
204.             try {
205.                 rootStorage =
206.                         CheckpointStorageLoader.load(
207.                                 applicationConfiguredStorage,
208.                                 null,
209.                                 rootBackend,
210.                                 jobManagerConfig,
211.                                 classLoader,
212.                                 log);
213.             } catch (IllegalConfigurationException | DynamicCodeLoadingException e) {
214.                 throw new JobExecutionException(
215.                         jobId, "Could not instantiate configured checkpoint storage", e);
216.             }
218.             // instantiate the user-defined checkpoint hooks
219.                         // 示例化用户自定义的cp hook
220.             final SerializedValue<MasterTriggerRestoreHook.Factory[]> serializedHooks =
221.                     snapshotSettings.getMasterHooks();
222.             final List<MasterTriggerRestoreHook<?>> hooks;
224.             if (serializedHooks == null) {
225.                 hooks = Collections.emptyList();
226.             } else {
227.                 final MasterTriggerRestoreHook.Factory[] hookFactories;
228.                 try {
229.                     hookFactories = serializedHooks.deserializeValue(classLoader);
230.                 } catch (IOException | ClassNotFoundException e) {
231.                     throw new JobExecutionException(
232.                             jobId, "Could not instantiate user-defined checkpoint hooks", e);
233.                 }
235.                 final Thread thread = Thread.currentThread();
236.                 final ClassLoader originalClassLoader = thread.getContextClassLoader();
237.                 thread.setContextClassLoader(classLoader);
239.                 try {
240.                     hooks = new ArrayList<>(hookFactories.length);
241.                     for (MasterTriggerRestoreHook.Factory factory : hookFactories) {
242.                         hooks.add(MasterHooks.wrapHook(factory.create(), classLoader));
243.                     }
244.                 } finally {
245.                     thread.setContextClassLoader(originalClassLoader);
246.                 }
247.             }
249.             final CheckpointCoordinatorConfiguration chkConfig =
250.                     snapshotSettings.getCheckpointCoordinatorConfiguration();
252.                         // 创建CheckpointCoordinator对象
253.             executionGraph.enableCheckpointing(
254.                     chkConfig,
255.                     hooks,
256.                     checkpointIdCounter,
257.                     completedCheckpointStore,
258.                     rootBackend,
259.                     rootStorage,
260.                     checkpointStatsTracker,
261.                     checkpointsCleaner);
262.         }
264.         // create all the metrics for the Execution Graph
265.                 // 添加metrics指标
266.         metrics.gauge(RestartTimeGauge.METRIC_NAME, new RestartTimeGauge(executionGraph));
267.         metrics.gauge(DownTimeGauge.METRIC_NAME, new DownTimeGauge(executionGraph));
268.         metrics.gauge(UpTimeGauge.METRIC_NAME, new UpTimeGauge(executionGraph));
270.         return executionGraph;
271.     }

复制代码

在这个方法里，会先创建一个 ExecutionGraph 对象，然后对 JobGraph 中的 JobVertex 列表做一下排序（先把有 source 节点的 JobVertex 放在最前面，然后开始遍历，只有当前 JobVertex 的前置节点都已经添加到集合后才华把当前 JobVertex 节点添加到集合中），最后通过过 attachJobGraph() 方法生成详细的ExecutionGraph。
在上面的代码中，最需要焦点关注的方法是：executionGraph.attachJobGraph(sortedTopology);。该方法是创建ExecutionGraph的焦点方法，包括了创建上面我们说的各种ExecutionGraph中涉及的对象，以及连接它们来形成ExecutionGraph拓扑结构。
接下来我们进入该方法来一探究竟。
生成ExecutionGraph：attachJobGraph

先来看下attachJobGraph方法的实现：

1. public void attachJobGraph(List<JobVertex> topologicallySorted) throws JobException {
3.     assertRunningInJobMasterMainThread();
5.     LOG.debug(
6.             "Attaching {} topologically sorted vertices to existing job graph with {} "
7.                     + "vertices and {} intermediate results.",
8.             topologicallySorted.size(),
9.             tasks.size(),
10.             intermediateResults.size());
12.     final long createTimestamp = System.currentTimeMillis();
14.         // 遍历排序好的拓扑JobVertex
15.     for (JobVertex jobVertex : topologicallySorted) {
17.         if (jobVertex.isInputVertex() && !jobVertex.isStoppable()) {
18.             this.isStoppable = false;
19.         }
21.                 // 获取节点并行度信息
22.         VertexParallelismInformation parallelismInfo =
23.                 parallelismStore.getParallelismInfo(jobVertex.getID());
25.         // create the execution job vertex and attach it to the graph
26.                 // 创建ExecutionJobVertex
27.         ExecutionJobVertex ejv =
28.                 new ExecutionJobVertex(
29.                         this,
30.                         jobVertex,
31.                         maxPriorAttemptsHistoryLength,
32.                         rpcTimeout,
33.                         createTimestamp,
34.                         parallelismInfo,
35.                         initialAttemptCounts.getAttemptCounts(jobVertex.getID()));
37.                 // 重要方法！！！
38.                 // 构建ExecutionGraph，连接上下游节点
39.         ejv.connectToPredecessors(this.intermediateResults);
41.         ExecutionJobVertex previousTask = this.tasks.putIfAbsent(jobVertex.getID(), ejv);
42.         if (previousTask != null) {
43.             throw new JobException(
44.                     String.format(
45.                             "Encountered two job vertices with ID %s : previous=[%s] / new=[%s]",
46.                             jobVertex.getID(), ejv, previousTask));
47.         }
48.                
49.                 // 遍历ExecutionJobVertex的输出IntermediateResult
50.         for (IntermediateResult res : ejv.getProducedDataSets()) {
51.             IntermediateResult previousDataSet =
52.                     this.intermediateResults.putIfAbsent(res.getId(), res);
53.             if (previousDataSet != null) {
54.                 throw new JobException(
55.                         String.format(
56.                                 "Encountered two intermediate data set with ID %s : previous=[%s] / new=[%s]",
57.                                 res.getId(), res, previousDataSet));
58.             }
59.         }
61.         this.verticesInCreationOrder.add(ejv);
62.         this.numVerticesTotal += ejv.getParallelism();
63.     }
65.    
66.         //将所有的执行顶点和结果分区注册到分布式资源管理系统中，以便能够进行分布式调度。
67. registerExecutionVerticesAndResultPartitions(this.verticesInCreationOrder);
69.     // the topology assigning should happen before notifying new vertices to failoverStrategy
70.         // 转换执行拓扑
71.     executionTopology = DefaultExecutionTopology.fromExecutionGraph(this);
73.     partitionGroupReleaseStrategy =
74.             
75. // 创建部分组释放策略的方法，依赖于当前的调度的拓扑结构，这决定了当何时释放特定的中间数据结果所需的策略。
76. partitionGroupReleaseStrategyFactory.createInstance(getSchedulingTopology());
77. }

复制代码

在上面attchGraph方法中，起首遍历输入的排序后的JobVertex列表，对每一个JobVertex:

• 判断是否停止： 对于单个 JobVertex，如果它是一个输入顶点且不可停止，则整个 Job 不可停止。这在流处理任务中是常见的，一些输入数据源大概无法停止（如Kafka）。
  
• 获取并行信息并创建实行的顶点： 根据JobVertex的ID，从parallelismStore中获取并行信息。利用这些信息创建ExecutionJobVertex实例，它代表运行在特定TaskManager上的taskId，可以是待调度、运行或已完成的。
  
• 判断新添加的顶点是否已经存在： 如果试图添加一个已经存在的顶点，这意味着存在程序错误，因为每个JobVertex应当有唯一的ID。这将抛出异常。
  
• 判断数据集是否已经存在： 同样。如果试图添加一个已经存在的IntermediateResult，这将抛出异常。
  
• 添加实行顶点到创建顺序列表和增加总的顶点数量： 记载创建顶点的顺序能够确保在实行时能够按照精确的依赖关系举行。并同时更新总的顶点数量。
  

遍历完成后， 注册实行顶点和结果分区，将所有的实行顶点和结果分区注册到分布式资源管理体系中，以便能够举行分布式调度。
利用DefaultExecutionTopology工具类将ExecutionGraph转换为SchedulingTopology，如许便于任务调度器举行处理。
最后，调用partitionGroupReleaseStrategyFactory.createInstance(getSchedulingTopology())根据当前的调度的拓扑结构来创建组开释策略，这决定了当何时开释特定的中心数据结果所需的策略。
上面流程中，最需要关注的方法就是new ExecutionJobVertex和ejv.connectToPredecessors(this.intermediateResults);
接下来，我们分别对其举行探究。
创建 ExecutionJobVertex 对象

进入到该方法的源码中：

1. @VisibleForTesting
2. public ExecutionJobVertex(
3.         InternalExecutionGraphAccessor graph,
4.         JobVertex jobVertex,
5.         int maxPriorAttemptsHistoryLength,
6.         Time timeout,
7.         long createTimestamp,
8.         VertexParallelismInformation parallelismInfo,
9.         SubtaskAttemptNumberStore initialAttemptCounts)
10.         throws JobException {
12.     if (graph == null || jobVertex == null) {
13.         throw new NullPointerException();
14.     }
16.     this.graph = graph;
17.     this.jobVertex = jobVertex;
19.     this.parallelismInfo = parallelismInfo;
21.     // verify that our parallelism is not higher than the maximum parallelism
22.     if (this.parallelismInfo.getParallelism() > this.parallelismInfo.getMaxParallelism()) {
23.         throw new JobException(
24.                 String.format(
25.                         "Vertex %s's parallelism (%s) is higher than the max parallelism (%s). Please lower the parallelism or increase the max parallelism.",
26.                         jobVertex.getName(),
27.                         this.parallelismInfo.getParallelism(),
28.                         this.parallelismInfo.getMaxParallelism()));
29.     }
31.     this.resourceProfile =
32.             ResourceProfile.fromResourceSpec(jobVertex.getMinResources(), MemorySize.ZERO);
34.     this.taskVertices = new ExecutionVertex[this.parallelismInfo.getParallelism()];
36.     this.inputs = new ArrayList<>(jobVertex.getInputs().size());
38.     // take the sharing group
39.     this.slotSharingGroup = checkNotNull(jobVertex.getSlotSharingGroup());
40.     this.coLocationGroup = jobVertex.getCoLocationGroup();
42.     // create the intermediate results
43.     this.producedDataSets =
44.             new IntermediateResult[jobVertex.getNumberOfProducedIntermediateDataSets()];
46.     for (int i = 0; i < jobVertex.getProducedDataSets().size(); i++) {
47.         final IntermediateDataSet result = jobVertex.getProducedDataSets().get(i);
49.         this.producedDataSets[i] =
50.                 new IntermediateResult(
51.                         result.getId(),
52.                         this,
53.                         this.parallelismInfo.getParallelism(),
54.                         result.getResultType());
55.     }
57.     // create all task vertices
58.     for (int i = 0; i < this.parallelismInfo.getParallelism(); i++) {
59.         ExecutionVertex vertex =
60.                 new ExecutionVertex(
61.                         this,
62.                         i,
63.                         producedDataSets,
64.                         timeout,
65.                         createTimestamp,
66.                         maxPriorAttemptsHistoryLength,
67.                         initialAttemptCounts.getAttemptCount(i));
69.         this.taskVertices[i] = vertex;
70.     }
72.     // sanity check for the double referencing between intermediate result partitions and
73.     // execution vertices
74.     for (IntermediateResult ir : this.producedDataSets) {
75.         if (ir.getNumberOfAssignedPartitions() != this.parallelismInfo.getParallelism()) {
76.             throw new RuntimeException(
77.                     "The intermediate result's partitions were not correctly assigned.");
78.         }
79.     }
81.     final List<SerializedValue<OperatorCoordinator.Provider>> coordinatorProviders =
82.             getJobVertex().getOperatorCoordinators();
83.     if (coordinatorProviders.isEmpty()) {
84.         this.operatorCoordinators = Collections.emptyList();
85.     } else {
86.         final ArrayList<OperatorCoordinatorHolder> coordinators =
87.                 new ArrayList<>(coordinatorProviders.size());
88.         try {
89.             for (final SerializedValue<OperatorCoordinator.Provider> provider :
90.                     coordinatorProviders) {
91.                 coordinators.add(
92.                         OperatorCoordinatorHolder.create(
93.                                 provider, this, graph.getUserClassLoader()));
94.             }
95.         } catch (Exception | LinkageError e) {
96.             IOUtils.closeAllQuietly(coordinators);
97.             throw new JobException(
98.                     "Cannot instantiate the coordinator for operator " + getName(), e);
99.         }
100.         this.operatorCoordinators = Collections.unmodifiableList(coordinators);
101.     }
103.     // set up the input splits, if the vertex has any
104.     try {
105.         @SuppressWarnings("unchecked")
106.         InputSplitSource<InputSplit> splitSource =
107.                 (InputSplitSource<InputSplit>) jobVertex.getInputSplitSource();
109.         if (splitSource != null) {
110.             Thread currentThread = Thread.currentThread();
111.             ClassLoader oldContextClassLoader = currentThread.getContextClassLoader();
112.             currentThread.setContextClassLoader(graph.getUserClassLoader());
113.             try {
114.                 inputSplits =
115.                         splitSource.createInputSplits(this.parallelismInfo.getParallelism());
117.                 if (inputSplits != null) {
118.                     splitAssigner = splitSource.getInputSplitAssigner(inputSplits);
119.                 }
120.             } finally {
121.                 currentThread.setContextClassLoader(oldContextClassLoader);
122.             }
123.         } else {
124.             inputSplits = null;
125.         }
126.     } catch (Throwable t) {
127.         throw new JobException(
128.                 "Creating the input splits caused an error: " + t.getMessage(), t);
129.     }
130. }

复制代码

在上面这段代码中，主要实现了ExecutionVertex的创建和IntermediateResult对象的创建：

• 遍历当前JobVertex的输出IntermediateDataSet列表，并根据IntermediateDataSet来创建相应的IntermediateResult对象。每个IntermediateDataSet都会对应一个IntermediateResult。
  
• 根据当前JobVertex的并发度，来创建雷同数量的ExecutionVertex对象，每个ExecutionVertex对象代表一个并行计算任务，在实际实行时就是一个Task任务。
  

创建ExecutionVertex对象

进一步地，我们观察创建ExecutionVertex对象的实现逻辑如下所示：

1. public ExecutionVertex(
2.         ExecutionJobVertex jobVertex,
3.         int subTaskIndex,
4.         IntermediateResult[] producedDataSets,
5.         Time timeout,
6.         long createTimestamp,
7.         int maxPriorExecutionHistoryLength,
8.         int initialAttemptCount) {
10.     this.jobVertex = jobVertex;
11.     this.subTaskIndex = subTaskIndex;
12.     this.executionVertexId = new ExecutionVertexID(jobVertex.getJobVertexId(), subTaskIndex);
13.     this.taskNameWithSubtask =
14.             String.format(
15.                     "%s (%d/%d)",
16.                     jobVertex.getJobVertex().getName(),
17.                     subTaskIndex + 1,
18.                     jobVertex.getParallelism());
20.     this.resultPartitions = new LinkedHashMap<>(producedDataSets.length, 1);
22.         // 根据IntermediateResult创建当前subTaskIndex分区下的IntermediateResultPartiton
23.     for (IntermediateResult result : producedDataSets) {
24.         IntermediateResultPartition irp =
25.                 new IntermediateResultPartition(
26.                         result,
27.                         this,
28.                         subTaskIndex,
29.                         getExecutionGraphAccessor().getEdgeManager());
30.                 // 记录当前分区的irp到ir中
31.         result.setPartition(subTaskIndex, irp);
32.                
33.                 // 记录分区ip与irp的对应关系
34.         resultPartitions.put(irp.getPartitionId(), irp);
35.     }
37.     this.priorExecutions = new EvictingBoundedList<>(maxPriorExecutionHistoryLength);
39.         // 创建对应的Execution对象，初始化时initialAttempCount为0，如果后面重新调度这个task，它会自增加1
40.     this.currentExecution =
41.             new Execution(
42.                     getExecutionGraphAccessor().getFutureExecutor(),
43.                     this,
44.                     initialAttemptCount,
45.                     createTimestamp,
46.                     timeout);
48.     getExecutionGraphAccessor().registerExecution(currentExecution);
50.     this.timeout = timeout;
51.     this.inputSplits = new ArrayList<>();
52. }

复制代码

上述创建ExecutionVertex的过程主要实现了以下步骤：

• 生成中心结果分区IntermediateResultPartition
  中心结果分区代表一个并行任务产生的输出，同一并行任务大概会有多个输出（对应多个后续任务），也就是多个中心结果分区。
  

• 基于 result，在相应的索引 subTaskIndex 上创建一个 IntermediateResultPartition 并给它赋值。IntermediateResultPartition 提供了并行任务的输出数据，对应于某个特定实行顶点 ExecutionVertex 的并行子任务。
  
• 在创建过程中，需要使用 getExecutionGraphAccessor().getEdgeManager() 获取边管理器，边管理器是用于维护这个分区与其它 ExecutionVertex 之间的连接关系。
  
• 记载这个 IntermediateResultPartition 到 result 中的相应索引位置，并在 resultPartitions 映射表中保存 IntermediateResultPartition。
  

• 创建实行（Execution）对象：
  这一过程是基于 Execution 的构造函数引发的。它用于代表该 ExecutionVertex 在某一特定点时间的一次尝试实行。创建 Execution 实例后，会将其注册到实行图（ExecutionGraph）中，以便于后续调度和实行任务。
  

通过以上流程，生成了中心结果分区，映射了每一个分区和其对应的任务关系，并且创建了 Execution 对象用于管理并跟踪任务的实行状态。
在创建好ExecutionVertex和IntermediateResultPartition后，根据上面的流程图，就是考虑如何将它们举行连接生成ExecutionGraph了。
这部分的实现逻辑就在attachJobGraph方法的ejv.connectToPredecessors(this.intermediateResults);方法中实现的。
生成ExecutionGraph

同样地，我们进入源码来深入观察一下实现逻辑：

1. public void connectToPredecessors(
2.         Map<IntermediateDataSetID, IntermediateResult> intermediateDataSets)
3.         throws JobException {
5.     List<JobEdge> inputs = jobVertex.getInputs();
7.     if (LOG.isDebugEnabled()) {
8.         LOG.debug(
9.                 String.format(
10.                         "Connecting ExecutionJobVertex %s (%s) to %d predecessors.",
11.                         jobVertex.getID(), jobVertex.getName(), inputs.size()));
12.     }
14.     for (int num = 0; num < inputs.size(); num++) {
15.         JobEdge edge = inputs.get(num);
17.         if (LOG.isDebugEnabled()) {
18.             if (edge.getSource() == null) {
19.                 LOG.debug(
20.                         String.format(
21.                                 "Connecting input %d of vertex %s (%s) to intermediate result referenced via ID %s.",
22.                                 num,
23.                                 jobVertex.getID(),
24.                                 jobVertex.getName(),
25.                                 edge.getSourceId()));
26.             } else {
27.                 LOG.debug(
28.                         String.format(
29.                                 "Connecting input %d of vertex %s (%s) to intermediate result referenced via predecessor %s (%s).",
30.                                 num,
31.                                 jobVertex.getID(),
32.                                 jobVertex.getName(),
33.                                 edge.getSource().getProducer().getID(),
34.                                 edge.getSource().getProducer().getName()));
35.             }
36.         }
38.         // fetch the intermediate result via ID. if it does not exist, then it either has not
39.         // been created, or the order
40.         // in which this method is called for the job vertices is not a topological order
41.         IntermediateResult ires = intermediateDataSets.get(edge.getSourceId());
42.         if (ires == null) {
43.             throw new JobException(
44.                     "Cannot connect this job graph to the previous graph. No previous intermediate result found for ID "
45.                             + edge.getSourceId());
46.         }
48.         this.inputs.add(ires);
50.         EdgeManagerBuildUtil.connectVertexToResult(this, ires, edge.getDistributionPattern());
51.     }
52. }

复制代码

这段代码主要完成了将当前的ExecutionJobVertex与其前置任务（predecessors）连接的流程。传入的参数intermediateDatasets包罗了JobGraph中所有的中心计算结果，这些结果是由上游前置任务产生的。
需要注意的是，该过程要求连接操作的实行顺序应遵循任务的拓扑顺序。Flink的计算任务通常由多个阶段构成，每个阶段的输出是下一个阶段的输入，每个阶段（JobVertex）都处理一种类型的计算，例如map或reduce。
流程大致如下：

• 获取输入： 起首获取jobVertex的输入，输入是JobEdge列表，每一条JobEdge都代表一个上游产生的中心数据集和连接上下游的方式（例如HASH, BROADCAST）。
  
• 循环处理每个输入： 然后遍历这些inputs，对于每一条JobEdge：
  
  
  
  • 基于edge.getSourceId()从intermediateDatasets获取IntermediateResult，这是一个中心计算结果。
    
  • 查抄该中心结果是否存在，如果不存在，则表示这不是一个拓扑排序，因为预期的情况是当你尝试访问一个中心结果时，它应该已经被创建了。如果找不到，抛出一个异常。
    
  • 如果存在（没有异常被抛出），将找到的IntermediateResult添加到ExecutionJobVertex的inputs（List类型）中，如许当前任务就知道它的输入来自哪些中心结果。
    
  • 调用EdgeManagerBuildUtil.connectVertexToResult方法来建立当前ExecutionJobVertex与找到的IntermediateResult之间的连接。 EdgeManager是Flink中负责管理输入输出边的组件，它表现地记载了发送端的分区和吸收端的分区对应关系。
    
  
  

这个流程紧张的是建立了Job中每个任务的实行依赖关系，并明确了数据传输的方式，让任务在实行时清楚本身的输入来自哪里，当任务实行完成后，它产生的输出会通过何种方式被发送到哪些任务。
详细的连接方式，我们需要继承进入到EdgeManagerBuildUtil.connectVertexToResult方法中。其源码如下所示：

1. /**
2. * Calculate the connections between {@link ExecutionJobVertex} and {@link IntermediateResult} *
3. * based on the {@link DistributionPattern}.
4. *
5. * @param vertex the downstream consumer {@link ExecutionJobVertex}
6. * @param intermediateResult the upstream consumed {@link IntermediateResult}
7. * @param distributionPattern the {@link DistributionPattern} of the edge that connects the
8. *     upstream {@link IntermediateResult} and the downstream {@link IntermediateResult}
9. */
10. static void connectVertexToResult(
11.         ExecutionJobVertex vertex,
12.         IntermediateResult intermediateResult,
13.         DistributionPattern distributionPattern) {
15.     switch (distributionPattern) {
16.                 // 点对点的连接方式
17.         case POINTWISE:
18.             connectPointwise(vertex.getTaskVertices(), intermediateResult);
19.             break;
20.                 // 全连接的方式
21.         case ALL_TO_ALL:
22.             connectAllToAll(vertex.getTaskVertices(), intermediateResult);
23.             break;
24.         default:
25.             throw new IllegalArgumentException("Unrecognized distribution pattern.");
26.     }
27. }

复制代码

会根据DistributionPattern选择不同的连接方式，这里主要分两种情况，DistributionPattern是跟Partitioner的配置有关。
这里以POINTWISE的连接方式来举例，看一下其是如安在构造ExecutionGraph时连接上下游节点的。

1. private static void connectPointwise(
2.         ExecutionVertex[] taskVertices, IntermediateResult intermediateResult) {
4.     final int sourceCount = intermediateResult.getPartitions().length;
5.     final int targetCount = taskVertices.length;
7.     if (sourceCount == targetCount) {
8.         for (int i = 0; i < sourceCount; i++) {
9.             ExecutionVertex executionVertex = taskVertices[i];
10.             IntermediateResultPartition partition = intermediateResult.getPartitions()[i];
12.             ConsumerVertexGroup consumerVertexGroup =
13.                     ConsumerVertexGroup.fromSingleVertex(executionVertex.getID());
14.             partition.addConsumers(consumerVertexGroup);
16.             ConsumedPartitionGroup consumedPartitionGroup =
17.                     createAndRegisterConsumedPartitionGroupToEdgeManager(
18.                             partition.getPartitionId(), intermediateResult);
19.             executionVertex.addConsumedPartitionGroup(consumedPartitionGroup);
20.         }
21.     } else if (sourceCount > targetCount) {
22.         for (int index = 0; index < targetCount; index++) {
24.             ExecutionVertex executionVertex = taskVertices[index];
25.             ConsumerVertexGroup consumerVertexGroup =
26.                     ConsumerVertexGroup.fromSingleVertex(executionVertex.getID());
28.             int start = index * sourceCount / targetCount;
29.             int end = (index + 1) * sourceCount / targetCount;
31.             List<IntermediateResultPartitionID> consumedPartitions =
32.                     new ArrayList<>(end - start);
34.             for (int i = start; i < end; i++) {
35.                 IntermediateResultPartition partition = intermediateResult.getPartitions()[i];
36.                 partition.addConsumers(consumerVertexGroup);
38.                 consumedPartitions.add(partition.getPartitionId());
39.             }
41.             ConsumedPartitionGroup consumedPartitionGroup =
42.                     createAndRegisterConsumedPartitionGroupToEdgeManager(
43.                             consumedPartitions, intermediateResult);
44.             executionVertex.addConsumedPartitionGroup(consumedPartitionGroup);
45.         }
46.     } else {
47.         for (int partitionNum = 0; partitionNum < sourceCount; partitionNum++) {
49.             IntermediateResultPartition partition =
50.                     intermediateResult.getPartitions()[partitionNum];
51.             ConsumedPartitionGroup consumedPartitionGroup =
52.                     createAndRegisterConsumedPartitionGroupToEdgeManager(
53.                             partition.getPartitionId(), intermediateResult);
55.             int start = (partitionNum * targetCount + sourceCount - 1) / sourceCount;
56.             int end = ((partitionNum + 1) * targetCount + sourceCount - 1) / sourceCount;
58.             List<ExecutionVertexID> consumers = new ArrayList<>(end - start);
60.             for (int i = start; i < end; i++) {
61.                 ExecutionVertex executionVertex = taskVertices[i];
62.                 executionVertex.addConsumedPartitionGroup(consumedPartitionGroup);
64.                 consumers.add(executionVertex.getID());
65.             }
67.             ConsumerVertexGroup consumerVertexGroup =
68.                     ConsumerVertexGroup.fromMultipleVertices(consumers);
69.             partition.addConsumers(consumerVertexGroup);
70.         }
71.     }
72. }

复制代码

上面这段代码的目标是通过“点对点”的方式（即每个生产者产生的数据只被一个消费者消费）建立任务节点（ExecutionVertex）与中心结果集（IntermediateResultPartition）之间的连接关系。
这个方法的逻辑主要是根据上游任务产生的IntermediateResultPartition的数量（源）和下游ExecutionVertex节点数量（目标）的比例关系，做不同的操作：

• 源和目标数量相等：方法会将每个源中心结果分区与对应的下游ExecutionVertex节点连接。这种情况下，每个任务都完全独立，只会消费一个特定的上游中心结果分区
  
• 源数量大于目标数量：源中心结果分区会被尽大概平均地分配给下游ExecutionVertex节点，即每个ExecutionVertex大概会消费多个源中心结果分区数据。
  
• 源数量小于目标数量：每个源中心结果分区大概会被分配给多个下游ExecutionVertex节点消费，即多个ExecutionVertex节点大概消费同一个源中心结果分区数据。
  

⠀在实行连接的过程中，会创建ConsumerVertexGroup和ConsumedPartitionGroup：

• ConsumerVertexGroup包罗一组吸收同一个中心结果分区（IntermediateResultPartition）的顶点集合。
  
• ConsumedPartitionGroup包罗顶点要消费的一组中心结果分区。
  

⠀注意，当源数量小于目标数量时，会有多个任务消费同一个源数据，以是需要使用ConsumerVertexGroup.fromMultipleVertices(consumers)来创建ConsumerVertexGroup。
几种连接情况的示例图如下所示：

到这里，这个作业的 ExecutionGraph 就创建完成了，有了 ExecutionGraph，JobManager 才华对这个作业做相应的调度。
总结

本文详细介绍了JobGraph生成ExecutionGraph的流程，介绍了ExecutionJobVertex、ExecutionVertex、IntermediateResult、IntermediateResultPartition相干概念的原理和生成过程。最后我们介绍了Flink在生成ExecutionGraph时是如何实现IntermediateResultPartition和ExecutionVertex的连接的。
到这里，StreamGraph、JobGraph和Execution的生成过程，在近来的三篇文章中都已经详细解说完成了，当然除了我们介绍的内容外，另有更多的实现细节没有介绍，有兴趣的读者可以参考文本来阅读源码，以此加深本身的明白和对更多实现细节的挖掘。
最后，再对StreamGraph、JobGraph和ExecutionGraph做一个总结：

• StreamGraph. StreamGraph 是表示 Flink 流计算的图模子，它是用户界说的计算逻辑的内部表示形式，是最原始的用户逻辑，一个没有做任何优化的DataFlow;
  
• JobGraph. JobGraph 由一个或多个 JobVertex 对象和它们之间的 JobEdge 对象构成，包罗并行任务的信息。在JobGraph中对StreamGraph举行了优化，将能够归并在同个算子链中的操作符举行归并，以此减少任务实行时的上下文切换，提任务实行性能。
  
• ExecutionGraph. ExecutionGraph是 JobGraph 的并发实行版本，由 ExecutionVertex 和 IntermediateResultPartition 构成。每个 JobVertex 会被转换为一个或多个 ExecutionVertex，ExecutorGraph 包罗了每个任务的全部实例，包罗它们的状态、位置、输入输出结果。ExecutionGraph 是 Flink 中最焦点的部分，它用于任务的调度、失败恢复等。
  

参考：
https://matt33.com/2019/12/20/flink-execution-graph-4/

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。