从头分析flink源码第六篇之ExecutionGraph的生成

    // 创建和恢复ExecutionGraph
        this.executionGraph = createAndRestoreExecutionGraph(jobManagerJobMetricGroup, checkNotNull(shuffleMaster), checkNotNull(partitionTracker), checkNotNull(executionDeploymentTracker), initializationTimestamp);
        // 调度拓朴里有partition信息和pipelineRegion信息
        this.schedulingTopology = executionGraph.getSchedulingTopology();

    private ExecutionGraph createAndRestoreExecutionGraph(
        JobManagerJobMetricGroup currentJobManagerJobMetricGroup,
        ShuffleMaster<?> shuffleMaster,
        JobMasterPartitionTracker partitionTracker,
        ExecutionDeploymentTracker executionDeploymentTracker,
        long initializationTimestamp) throws Exception {
        ExecutionGraph newExecutionGraph = createExecutionGraph(currentJobManagerJobMetricGroup, shuffleMaster, partitionTracker, executionDeploymentTracker, initializationTimestamp);
        // checkpoint 协调器
        final CheckpointCoordinator checkpointCoordinator = newExecutionGraph.getCheckpointCoordinator();
        if (checkpointCoordinator != null) {
            // check whether we find a valid checkpoint
            if (!checkpointCoordinator.restoreInitialCheckpointIfPresent(
                    new HashSet<>(newExecutionGraph.getAllVertices().values()))) {
                // check whether we can restore from a savepoint
                tryRestoreExecutionGraphFromSavepoint(newExecutionGraph, jobGraph.getSavepointRestoreSettings());
            }
        }
        return newExecutionGraph;
    }

private ExecutionGraph createExecutionGraph(
        JobManagerJobMetricGroup currentJobManagerJobMetricGroup,
        ShuffleMaster<?> shuffleMaster,
        final JobMasterPartitionTracker partitionTracker,
        ExecutionDeploymentTracker executionDeploymentTracker,
        long initializationTimestamp) throws JobExecutionException, JobException {
        // 任务布署监听器，用于记录布署过程中发生的动作
        ExecutionDeploymentListener executionDeploymentListener = new ExecutionDeploymentTrackerDeploymentListenerAdapter(executionDeploymentTracker);
        // 执行状态更新监听器
        ExecutionStateUpdateListener executionStateUpdateListener = (execution, newState) -> {
            if (newState.isTerminal()) {
                executionDeploymentTracker.stopTrackingDeploymentOf(execution);
            }
        };
        // 任务失败容错策略
        final FailoverStrategy.Factory failoverStrategy = legacyScheduling ?
            FailoverStrategyLoader.loadFailoverStrategy(jobMasterConfiguration, log) :
            // 不做处理
            new NoOpFailoverStrategy.Factory();
        // 构造ExecutionGraph
        return ExecutionGraphBuilder.buildGraph(
            null,
            jobGraph,
            jobMasterConfiguration,
            futureExecutor,
            ioExecutor,
            slotProvider,
            userCodeLoader,
            checkpointRecoveryFactory,
            rpcTimeout,
            restartStrategy,
            currentJobManagerJobMetricGroup,
            blobWriter,
            slotRequestTimeout,
            log,
            shuffleMaster,
            partitionTracker,
            failoverStrategy,
            executionDeploymentListener,
            executionStateUpdateListener,
            initializationTimestamp);
    }

JobManagerJobMetricGroupFactory jobMetricGroupFactory

final ShuffleMaster<?> shuffleMaster = ShuffleServiceLoader.loadShuffleServiceFactory(configuration).createShuffleMaster(configuration);

    @Override
    public NettyShuffleMaster createShuffleMaster(Configuration configuration) {
        return NettyShuffleMaster.INSTANCE;
    }

    this.partitionTracker = checkNotNull(partitionTrackerFactory)
            .create(resourceID -> {
                Tuple2<TaskManagerLocation, TaskExecutorGateway> taskManagerInfo = registeredTaskManagers.get(resourceID);
                if (taskManagerInfo == null) {
                    return Optional.empty();
                }
                return Optional.of(taskManagerInfo.f1);
            });

        final String jobName = jobGraph.getName();
        final JobID jobId = jobGraph.getJobID();
        // jobInformation中包括序列化配置，job配置，用户jar包在blobStore中的blobKey，classpath信息等
        final JobInformation jobInformation = new JobInformation(
            jobId,jobName,
            jobGraph.getSerializedExecutionConfig(),jobGraph.getJobConfiguration(),
            jobGraph.getUserJarBlobKeys(),jobGraph.getClasspaths());
        // 历史上保留的最大执行尝试数
        final int maxPriorAttemptsHistoryLength =
                jobManagerConfig.getInteger(JobManagerOptions.MAX_ATTEMPTS_HISTORY_SIZE);
        // intermediate result partitions释放策略工厂，决定intermediate result partitions何时释放
        final PartitionReleaseStrategy.Factory partitionReleaseStrategyFactory =
        PartitionReleaseStrategyFactoryLoader.loadPartitionReleaseStrategyFactory(jobManagerConfig);
        // create a new execution graph, if none exists so far
        final ExecutionGraph executionGraph;
        try {
            // 创建executionGraph
            executionGraph = (prior != null) ? prior :
                new ExecutionGraph(
                    jobInformation,futureExecutor,ioExecutor,rpcTimeout,restartStrategy,
                    maxPriorAttemptsHistoryLength,failoverStrategyFactory,slotProvider,classLoader,
                    blobWriter,allocationTimeout,partitionReleaseStrategyFactory,shuffleMaster,
                    partitionTracker,jobGraph.getScheduleMode(),executionDeploymentListener,
                    executionStateUpdateListener,initializationTimestamp);
        } catch (IOException e) {
            throw new JobException("Could not create the ExecutionGraph.", e);
        }

    /** All job vertices that are part of this graph. */
    private final Map<JobVertexID, ExecutionJobVertex> tasks;

    /** All vertices, in the order in which they were created. **/
    private final List<ExecutionJobVertex> verticesInCreationOrder;

    /** All intermediate results that are part of this graph. */
    private final Map<IntermediateDataSetID, IntermediateResult> intermediateResults;

    /** The currently executed tasks, for callbacks. */
    private final Map<ExecutionAttemptID, Execution> currentExecutions;

executionGraph.setJsonPlan(JsonPlanGenerator.generatePlan(jobGraph));

    for (JobVertex vertex : jobGraph.getVertices()) {
            String executableClass = vertex.getInvokableClassName();
            if (executableClass == null || executableClass.isEmpty()) {
                throw new JobSubmissionException(jobId,
                        "The vertex " + vertex.getID() + " (" + vertex.getName() + ") has no invokable class.");
            }
            try {
                vertex.initializeOnMaster(classLoader);
            }

// topologically sort the job vertices and attach the graph to the existing one
// jobGraph的taskVertices为map结构，里面都是无序的，这个方法的作用是将这些无序的转成有序列表
List<JobVertex> sortedTopology = jobGraph.getVerticesSortedTopologicallyFromSources();
if (log.isDebugEnabled()) {
    log.debug("Adding {} vertices from job graph {} ({}).", sortedTopology.size(), jobName, jobId);
}
// 创建ExecutionJobVertex和IntermediateResult
executionGraph.attachJobGraph(sortedTopology);

    public void attachJobGraph(List<JobVertex> topologiallySorted) throws JobException {
        assertRunningInJobMasterMainThread();
        LOG.debug(
            "Attaching {} topologically sorted vertices to existing job graph with {} " +
                "vertices and {} intermediate results.",
            topologiallySorted.size(),
            tasks.size(),
            intermediateResults.size());
        final ArrayList<ExecutionJobVertex> newExecJobVertices = new ArrayList<>(topologiallySorted.size());
        final long createTimestamp = System.currentTimeMillis();
        for (JobVertex jobVertex : topologiallySorted) {
            if (jobVertex.isInputVertex() && !jobVertex.isStoppable()) {
                this.isStoppable = false;
            }
            // 将jobGraph的顶点转化成ExecutionJobVertex
            // create the execution job vertex and attach it to the graph
            ExecutionJobVertex ejv = new ExecutionJobVertex(
                this,
                jobVertex,
                1,
                maxPriorAttemptsHistoryLength,
                rpcTimeout,
                globalModVersion,
                createTimestamp);
            // 连接ExecutionJobVertex与intermediateResult之间，all_to_all point wise  创建ExecuteEdge信息并与上下游节点建立连接
            ejv.connectToPredecessors(this.intermediateResults);
            ExecutionJobVertex previousTask = this.tasks.putIfAbsent(jobVertex.getID(), ejv);
            if (previousTask != null) {
                throw new JobException(String.format(
                    "Encountered two job vertices with ID %s : previous=[%s] / new=[%s]",
                    jobVertex.getID(),
                    ejv,
                    previousTask));
            }
            for (IntermediateResult res : ejv.getProducedDataSets()) {
                IntermediateResult previousDataSet = this.intermediateResults.putIfAbsent(
                    res.getId(),
                    res);
                if (previousDataSet != null) {
                    throw new JobException(String.format(
                        "Encountered two intermediate data set with ID %s : previous=[%s] / new=[%s]",
                        res.getId(),
                        res,
                        previousDataSet));
                }
            }
            this.verticesInCreationOrder.add(ejv);
            this.numVerticesTotal += ejv.getParallelism();
            newExecJobVertices.add(ejv);
        }
        // 创建执行拓扑，并指定上下游的生产和消费partition
        // the topology assigning should happen before notifying new vertices to failoverStrategy
        executionTopology = DefaultExecutionTopology.fromExecutionGraph(this);
        failoverStrategy.notifyNewVertices(newExecJobVertices);
        partitionReleaseStrategy = partitionReleaseStrategyFactory.createInstance(
            getSchedulingTopology());
    }

// 将jobGraph的顶点转化成ExecutionJobVertex
            // create the execution job vertex and attach it to the graph
            ExecutionJobVertex ejv = new ExecutionJobVertex(
                this,
                jobVertex,
                1,
                maxPriorAttemptsHistoryLength,
                rpcTimeout,
                globalModVersion,
                createTimestamp);
            // 连接ExecutionJobVertex与intermediateResult之间，all_to_all point wise  创建ExecuteEdge信息并与上下游节点建立连接
            ejv.connectToPredecessors(this.intermediateResults);
            ExecutionJobVertex previousTask = this.tasks.putIfAbsent(jobVertex.getID(), ejv);
            if (previousTask != null) {
                throw new JobException(String.format(---------));
            }
            for (IntermediateResult res : ejv.getProducedDataSets()) {
                IntermediateResult previousDataSet = this.intermediateResults.putIfAbsent(
                    res.getId(),
                    res);
                if (previousDataSet != null) {
                    throw new JobException(---)
                }
            }
            this.verticesInCreationOrder.add(ejv);
            this.numVerticesTotal += ejv.getParallelism();
            newExecJobVertices.add(ejv);

   // 获取jobVertex的并行度，如果没有设置则会使用默认的并行度
           int vertexParallelism = jobVertex.getParallelism();
           // 确定subTask的数量
           int numTaskVertices = vertexParallelism > 0 ? vertexParallelism : defaultParallelism;
           // 最大并行度
           final int configuredMaxParallelism = jobVertex.getMaxParallelism();

           this.maxParallelismConfigured = (VALUE_NOT_SET != configuredMaxParallelism);
           // 如果jobVertex没有配置最大并行度，则会计算一个默认值
           // if no max parallelism was configured by the user, we calculate and set a default
           setMaxParallelismInternal(maxParallelismConfigured ?
                   configuredMaxParallelism : KeyGroupRangeAssignment.computeDefaultMaxParallelism(numTaskVertices));

     // 当前节点产生的中间结果，可能它的下游对应多个分区（一对多）
           // create the intermediate results
           this.producedDataSets = new IntermediateResult[jobVertex.getNumberOfProducedIntermediateDataSets()];
           for (int i = 0; i < jobVertex.getProducedDataSets().size(); i++) {
               final IntermediateDataSet result = jobVertex.getProducedDataSets().get(i);
               // 当前节点产生中间数据的列表
               this.producedDataSets[i] = new IntermediateResult(
                       result.getId(),
                       this,
                       numTaskVertices,
                       result.getResultType());
           }

      // create all task vertices
              for (int i = 0; i < numTaskVertices; i++) {
                  // 根据并行度创建多个ExecutionVertex，每个ExecutionVertex对应一个subTask。多个subTask共用producedDataSets
                  ExecutionVertex vertex = new ExecutionVertex(
                          this,
                          i,
                          producedDataSets,
                          timeout,
                          initialGlobalModVersion,
                          createTimestamp,
                          maxPriorAttemptsHistoryLength);
                  this.taskVertices[i] = vertex;
              }

         this.resultPartitions = new LinkedHashMap<>(producedDataSets.length, 1);
                 for (IntermediateResult result : producedDataSets) {
                     // 中间结果分区，每个IntermediateResult对应一个IntermediateResultPartition，然后每个ExecutionVertex对象对应不同的subTaskIndex，
                     // 每个subTaskIndex都会与producedDataSets中所有的IntermediateResult对应的一个IntermediateResultPartition形成映射
                     IntermediateResultPartition irp = new IntermediateResultPartition(result, this, subTaskIndex);
                     // 在IntermediateResult中有一个partitions数组
                     result.setPartition(subTaskIndex, irp);
                     resultPartitions.put(irp.getPartitionId(), irp);
                 }

      this.inputEdges = new ExecutionEdge[jobVertex.getJobVertex().getInputs().size()][];

// 创建execution，一个顶点的一次执行。虽然{@link ExecutionVertex}可以执行多次(用于恢复、重新计算、重新配置)，但该类跟踪顶点和资源的单个执行状态。
        this.currentExecution = new Execution(
            getExecutionGraph().getFutureExecutor(),
            this,
            0,
            initialGlobalModVersion,
            createTimestamp,
            timeout);

// 创建执行拓扑，并指定上下游的生产和消费partition
// the topology assigning should happen before notifying new vertices to failoverStrategy
executionTopology = DefaultExecutionTopology.fromExecutionGraph(this);

    public static DefaultExecutionTopology fromExecutionGraph(ExecutionGraph executionGraph) {
        // 默认的执行拓朴
        final DefaultExecutionTopology topology = new DefaultExecutionTopology(executionGraph);
        // 生成流水线regions
        final List<DefaultSchedulingPipelinedRegion> pipelinedRegions = generatePipelinedRegions(topology);
        topology.setPipelinedRegions(pipelinedRegions);
        ensureCoLocatedVerticesInSameRegion(pipelinedRegions, executionGraph);
        return topology;
    }