Flink源码分析系列文档目录

请点击：Flink 源码分析系列文档目录

从JobGraph到ExecutionGraph

JobGraph通过Dispatcher.submitJob方法提交。这是后续流程的入口方法。该方法调用了Dispatcher.internalSubmitJob，然后是Dispatcher.persistAndRunJob。

Dispatcher.persistAndRunJob方法存储并执行作业。如下所示：

private void persistAndRunJob(JobGraph jobGraph) throws Exception {
    jobGraphWriter.putJobGraph(jobGraph);
    runJob(jobGraph, ExecutionType.SUBMISSION);
}

Dispatcher.runJob接收JobGraph和执行类型两个参数。执行类型有两种：提交任务（SUBMISSION）和恢复任务（RECOVERY）。

private void runJob(JobGraph jobGraph, ExecutionType executionType) {
    // 确保JobID对应的这个作业目前不在运行状态，避免重复提交
    Preconditions.checkState(!runningJobs.containsKey(jobGraph.getJobID()));
    // 获取启动时时间戳
    long initializationTimestamp = System.currentTimeMillis();
    // 这里将JobManagerRunner创建出来
    // JobManagerRunner接下来会构造出JobManager
    CompletableFuture<JobManagerRunner> jobManagerRunnerFuture =
        createJobManagerRunner(jobGraph, initializationTimestamp);

    // 包装JobGraph相关信息供Dispatcher使用
    DispatcherJob dispatcherJob =
        DispatcherJob.createFor(
        jobManagerRunnerFuture,
        jobGraph.getJobID(),
        jobGraph.getName(),
        initializationTimestamp);
    
    // 将当前作业的ID加入runningJob集合
    // 表示当前作业已处于运行状态
    runningJobs.put(jobGraph.getJobID(), dispatcherJob);

    final JobID jobId = jobGraph.getJobID();

    // 处理Job派发结果
    final CompletableFuture<CleanupJobState> cleanupJobStateFuture =
        dispatcherJob
        .getResultFuture()
        .handleAsync(
        (dispatcherJobResult, throwable) -> {
            Preconditions.checkState(
                runningJobs.get(jobId) == dispatcherJob,
                "The job entry in runningJobs must be bound to the lifetime of the DispatcherJob.");

            if (dispatcherJobResult != null) {
                return handleDispatcherJobResult(
                    jobId, dispatcherJobResult, executionType);
            } else {
                return dispatcherJobFailed(jobId, throwable);
            }
        },
        getMainThreadExecutor());

    // 作业停止的时候，将JobID从runningJob中移除
    final CompletableFuture<Void> jobTerminationFuture =
        cleanupJobStateFuture
        .thenApply(cleanupJobState -> removeJob(jobId, cleanupJobState))
        .thenCompose(Function.identity());

    // 将作业ID和对应的作业停止future加入到dispatcherJobTerminationFutures集合维护
    FutureUtils.assertNoException(jobTerminationFuture);
    registerDispatcherJobTerminationFuture(jobId, jobTerminationFuture);
}

接下来是Dispatcher.createJobManagerRunner方法。

JobManager在Dispatcher中被创建出来，然后启动。创建JobManager的逻辑在createJobManagerRunner方法中，如下所示：

CompletableFuture<JobManagerRunner> createJobManagerRunner(
    JobGraph jobGraph, long initializationTimestamp) {
    final RpcService rpcService = getRpcService();
    return CompletableFuture.supplyAsync(
        () -> {
            try {
                // 使用工厂类创建JobManager
                // 传入了JobGraph和高可用服务
                JobManagerRunner runner =
                    jobManagerRunnerFactory.createJobManagerRunner(
                    jobGraph,
                    configuration,
                    rpcService,
                    highAvailabilityServices,
                    heartbeatServices,
                    jobManagerSharedServices,
                    new DefaultJobManagerJobMetricGroupFactory(
                        jobManagerMetricGroup),
                    fatalErrorHandler,
                    initializationTimestamp);
                // 启动JobManager
                // 实际上为启动JobManager的leader选举服务，选出JM主节点
                runner.start();
                return runner;
            } catch (Exception e) {
                throw new CompletionException(
                    new JobInitializationException(
                        jobGraph.getJobID(),
                        "Could not instantiate JobManager.",
                        e));
            }
        },
        ioExecutor); // do not use main thread executor. Otherwise, Dispatcher is blocked on
    // JobManager creation
}

此时，JobManager开始进行leader竞选活动。为了确保JobManager不存在单点故障问题，Flink设计了JobManager 高可用，可以同时运行多个JobManager实例。在Standalone部署方式中，JobManager的竞选通过Zookeeper来实现。Yarn集群模式下则通过Yarn的ApplicationMaster失败后自动重启动方式来确保JobManager的高可用。有关leader选举的内容请参见Flink 源码之leader选举（Zookeeper方式）。

一旦leader JM被选举出来，选举服务会调用对应JM的grantLeadership方法。该方法内容如下所示：

@Override
public void grantLeadership(final UUID leaderSessionID) {
    synchronized (lock) {
        if (shutdown) {
            log.debug(
                "JobManagerRunner cannot be granted leadership because it is already shut down.");
            return;
        }

        leadershipOperation =
            leadershipOperation.thenRun(
            ThrowingRunnable.unchecked(
                () -> {
                    synchronized (lock) {
                        // 主要逻辑是这个
                        // 检查作业调度状态并启动JobManager
                        verifyJobSchedulingStatusAndStartJobManager(
                            leaderSessionID);
                    }
                }));

        handleException(leadershipOperation, "Could not start the job manager.");
    }
}

接着我们跟踪到verifyJobSchedulingStatusAndStartJobManager方法。

@GuardedBy("lock")
private void verifyJobSchedulingStatusAndStartJobManager(UUID leaderSessionId)
    throws FlinkException {
    // 如果JobManager已停止，直接返回
    if (shutdown) {
        log.debug("Ignoring starting JobMaster because JobManagerRunner is already shut down.");
        return;
    }

    // 从JobRegistry中获取Job调度状态
    final RunningJobsRegistry.JobSchedulingStatus jobSchedulingStatus =
        getJobSchedulingStatus();

    // 如果作业已执行完毕
    // 调用作业执行完毕逻辑（实际上是作业未被当前JobManager完成运行的逻辑）
    if (jobSchedulingStatus == RunningJobsRegistry.JobSchedulingStatus.DONE) {
        jobAlreadyDone();
    } else {
        // 启动JobMaster
        startJobMaster(leaderSessionId);
    }
}

现在逻辑流转到了JobManagerRunnerImpl.startJobMaster方法。

该方法启动JobMaster。注册JobGraph，启动JobMaster服务并确认该JobMaster为leader。

@GuardedBy("lock")
private void startJobMaster(UUID leaderSessionId) throws FlinkException {
    log.info(
        "JobManager runner for job {} ({}) was granted leadership with session id {}.",
        jobGraph.getName(),
        jobGraph.getJobID(),
        leaderSessionId);

    try {
        // 注册JobGraph
        // 根据集群部署形式（Standalone，Zookeeper或K8s），采用不同的方式存储JobID
        runningJobsRegistry.setJobRunning(jobGraph.getJobID());
    } catch (IOException e) {
        throw new FlinkException(
            String.format(
                "Failed to set the job %s to running in the running jobs registry.",
                jobGraph.getJobID()),
            e);
    }

    // 启动JobMaster服务
    startJobMasterServiceSafely(leaderSessionId);

    // 确认该JobMaster是leader状态
    if (jobMasterService != null) {
        confirmLeaderSessionIdIfStillLeader(jobMasterService, leaderSessionId);
    }
}

JobManagerRunnerImpl.startJobMasterServiceSafely紧接着通过

DefaultJobMasterServiceFactory.createJobMasterService方法，创建出JobMaster并启动他的Rpc通信服务。

接下来。在JobMaster构造函数中存在构建Flink作业任务调度器的逻辑。JobMaster.createScheduler方法调用了

DefaultSlotPoolServiceSchedulerFactory.createScheduler创建Flink的调度器。该方法又调用了Scheduler工厂类的创建Scheduler实例这个方法DefaultSchedulerFactory.createInstance。

接下来的流程到了DefaultScheduler中。DefaultScheduler是Flink作业调度器的默认实现。它继承了SchedulerBase，SchedulerBase又实现了SchedulerNG接口。

SchedulerBase构造函数中调用了createAndRestoreExecutionGraph方法。

SchedulerBase.createAndRestoreExecutionGraph代码如下所示：

private ExecutionGraph createAndRestoreExecutionGraph(
    JobManagerJobMetricGroup currentJobManagerJobMetricGroup,
    CompletedCheckpointStore completedCheckpointStore,
    CheckpointsCleaner checkpointsCleaner,
    CheckpointIDCounter checkpointIdCounter,
    ShuffleMaster<?> shuffleMaster,
    JobMasterPartitionTracker partitionTracker,
    ExecutionDeploymentTracker executionDeploymentTracker,
    long initializationTimestamp,
    ComponentMainThreadExecutor mainThreadExecutor,
    JobStatusListener jobStatusListener)
    throws Exception {

    // 创建ExecutionGraph
    ExecutionGraph newExecutionGraph =
        createExecutionGraph(
        currentJobManagerJobMetricGroup,
        completedCheckpointStore,
        checkpointsCleaner,
        checkpointIdCounter,
        shuffleMaster,
        partitionTracker,
        executionDeploymentTracker,
        initializationTimestamp);

    // 获取ExecutionGraph中创建出的CheckpointCoordinator
    // 创建CheckpointCoordinator的过程后面章节有说明
    final CheckpointCoordinator checkpointCoordinator =
        newExecutionGraph.getCheckpointCoordinator();

    if (checkpointCoordinator != null) {
        // check whether we find a valid checkpoint
        // 检查是否存在一个最近的checkpoint
        if (!checkpointCoordinator.restoreInitialCheckpointIfPresent(
            new HashSet<>(newExecutionGraph.getAllVertices().values()))) {

            // check whether we can restore from a savepoint
            // 如果有，尝试从这个检查点恢复
            tryRestoreExecutionGraphFromSavepoint(
                newExecutionGraph, jobGraph.getSavepointRestoreSettings());
        }
    }

    // 设置任务失败监听器
    newExecutionGraph.setInternalTaskFailuresListener(
        new UpdateSchedulerNgOnInternalFailuresListener(this));
    // 设置作业状态监听器
    newExecutionGraph.registerJobStatusListener(jobStatusListener);
    // 设置JobMaster的主线程ThreadExecutor
    newExecutionGraph.start(mainThreadExecutor);

    return newExecutionGraph;
}

SchedulerBase.createExecutionGraph方法调用DefaultExecutionGraphBuilder，创建出ExecutionGraph。代码如下所示：

private ExecutionGraph createExecutionGraph(
    JobManagerJobMetricGroup currentJobManagerJobMetricGroup,
    CompletedCheckpointStore completedCheckpointStore,
    CheckpointsCleaner checkpointsCleaner,
    CheckpointIDCounter checkpointIdCounter,
    ShuffleMaster<?> shuffleMaster,
    final JobMasterPartitionTracker partitionTracker,
    ExecutionDeploymentTracker executionDeploymentTracker,
    long initializationTimestamp)
    throws JobExecutionException, JobException {

    // 创建Execution部署监听器
    ExecutionDeploymentListener executionDeploymentListener =
        new ExecutionDeploymentTrackerDeploymentListenerAdapter(executionDeploymentTracker);
    //创建Execution状态更新监听器
    ExecutionStateUpdateListener executionStateUpdateListener =
        (execution, newState) -> {
        if (newState.isTerminal()) {
            executionDeploymentTracker.stopTrackingDeploymentOf(execution);
        }
    };

    // 创建ExecutionGraph
    return DefaultExecutionGraphBuilder.buildGraph(
        jobGraph,
        jobMasterConfiguration,
        futureExecutor,
        ioExecutor,
        userCodeLoader,
        completedCheckpointStore,
        checkpointsCleaner,
        checkpointIdCounter,
        rpcTimeout,
        currentJobManagerJobMetricGroup,
        blobWriter,
        log,
        shuffleMaster,
        partitionTracker,
        TaskDeploymentDescriptorFactory.PartitionLocationConstraint.fromJobType(
            jobGraph.getJobType()),
        executionDeploymentListener,
        executionStateUpdateListener,
        initializationTimestamp,
        new DefaultVertexAttemptNumberStore());
}

ExecutionGraph相关概念

ExecutionGraph为Flink作业的物理执行计划。用来协调数据流的分布式执行过程。

和StreamGraph，JobGraph不同的是，ExecutionGraph是在JobManager中生成。

从ExecutionGraph也有顶点（Vertex）的概念，ExecutionGraph中的vertex为ExecutionJobVertex，和JobGraph中的JobVertex对应。从ExecutionGraph到JobGraph的过程中加入了并行度的概念，ExecutionJobVertex包含了与之对应的JobVertex中所有的并行任务。ExecutionJobVertex之中每一个并行的任务由ExecutionVertex代表。也就是说一个ExecutionJobVertex具有多少并行度，它下面就包含多少个ExecutionVertex。

ExecutionVertex可以被执行一次或多次（由于任务恢复，重计算或更新配置）ExecutionVertex的每一次执行都会生成一个Execution对象。Execution负责跟踪ExecutionVertex的任务执行状态变化和资源使用状况。

IntermediateResult和JobGraph中JobVertex的IntermediateDataSet的概念对应，用于表示两个相邻的ExecutionJobVertex之间数据传输过程中的临时存放点。IntermediateResult在ExecutionJobVertex创建的时候被构建出来，数量和该vertex的并行度一致。

DefaultExecutionGraphBuilder的buildGraph方法

public static ExecutionGraph buildGraph(
    JobGraph jobGraph,
    Configuration jobManagerConfig,
    ScheduledExecutorService futureExecutor,
    Executor ioExecutor,
    ClassLoader classLoader,
    CompletedCheckpointStore completedCheckpointStore,
    CheckpointsCleaner checkpointsCleaner,
    CheckpointIDCounter checkpointIdCounter,
    Time rpcTimeout,
    MetricGroup metrics,
    BlobWriter blobWriter,
    Logger log,
    ShuffleMaster<?> shuffleMaster,
    JobMasterPartitionTracker partitionTracker,
    TaskDeploymentDescriptorFactory.PartitionLocationConstraint partitionLocationConstraint,
    ExecutionDeploymentListener executionDeploymentListener,
    ExecutionStateUpdateListener executionStateUpdateListener,
    long initializationTimestamp,
    VertexAttemptNumberStore vertexAttemptNumberStore)
    throws JobExecutionException, JobException {

    checkNotNull(jobGraph, "job graph cannot be null");

    // 获取作业名称和作业ID
    final String jobName = jobGraph.getName();
    final JobID jobId = jobGraph.getJobID();

    // 创建JobInformation
    // JobInformation为ExecutionGraph中的job相关配置信息的封装类
    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);

    // 获取IntermediateResultPartition释放策略工厂类
    final PartitionReleaseStrategy.Factory partitionReleaseStrategyFactory =
        PartitionReleaseStrategyFactoryLoader.loadPartitionReleaseStrategyFactory(
        jobManagerConfig);

    // create a new execution graph, if none exists so far
    // 创建ExecutionGraph，后面章节分析
    final DefaultExecutionGraph executionGraph;
    try {
            executionGraph =
                    new DefaultExecutionGraph(
                            jobInformation,
                            futureExecutor,
                            ioExecutor,
                            rpcTimeout,
                            maxPriorAttemptsHistoryLength,
                            classLoader,
                            blobWriter,
                            partitionReleaseStrategyFactory,
                            shuffleMaster,
                            partitionTracker,
                            partitionLocationConstraint,
                            executionDeploymentListener,
                            executionStateUpdateListener,
                            initializationTimestamp,
                            vertexAttemptNumberStore);
    } catch (IOException e) {
        throw new JobException("Could not create the ExecutionGraph.", e);
    }

    // set the basic properties

    try {
        // 设置json格式的执行计划
        executionGraph.setJsonPlan(JsonPlanGenerator.generatePlan(jobGraph));
    } catch (Throwable t) {
        log.warn("Cannot create JSON plan for job", t);
        // give the graph an empty plan
        // 如果根据jobGraph生成json执行计划失败，设置一个空的执行计划
        executionGraph.setJsonPlan("{}");
    }

    // initialize the vertices that have a master initialization hook
    // file output formats create directories here, input formats create splits

    final long initMasterStart = System.nanoTime();
    log.info("Running initialization on master for job {} ({}).", jobName, jobId);

    for (JobVertex vertex : jobGraph.getVertices()) {
        // 获取节点调用的类名，即节点的task
        String executableClass = vertex.getInvokableClassName();
        // 确保每个节点的调用类必须存在
        if (executableClass == null || executableClass.isEmpty()) {
            throw new JobSubmissionException(
                jobId,
                "The vertex "
                + vertex.getID()
                + " ("
                + vertex.getName()
                + ") has no invokable class.");
        }

        try {
            // 根据不同的节点类型，调用job启动时节点的任务逻辑
            vertex.initializeOnMaster(classLoader);
        } catch (Throwable t) {
            throw new JobExecutionException(
                jobId,
                "Cannot initialize task '" + vertex.getName() + "': " + t.getMessage(),
                t);
        }
    }

    log.info(
        "Successfully ran initialization on master in {} ms.",
        (System.nanoTime() - initMasterStart) / 1_000_000);

    // topologically sort the job vertices and attach the graph to the existing one
    // 按照拓扑结构（数据流的顺序）排序，获取所有的Job顶点
    List<JobVertex> sortedTopology = jobGraph.getVerticesSortedTopologicallyFromSources();
    if (log.isDebugEnabled()) {
        log.debug(
            "Adding {} vertices from job graph {} ({}).",
            sortedTopology.size(),
            jobName,
            jobId);
    }
    // executionGraph绑定所有的Job节点
    executionGraph.attachJobGraph(sortedTopology);

    if (log.isDebugEnabled()) {
        log.debug(
            "Successfully created execution graph from job graph {} ({}).", jobName, jobId);
    }

    // configure the state checkpointing
    // 配置checkpoint
    // 如果启用了checkpoint
    if (isCheckpointingEnabled(jobGraph)) {
        // 从JobGraph获取checkpoint的配置
        // snapshotSettings的配置位于StreamingJobGraphGenerator
        JobCheckpointingSettings snapshotSettings = jobGraph.getCheckpointingSettings();
        
        // 获取所有触发checkpoint的顶点，即所有的数据输入顶点
        List<ExecutionJobVertex> triggerVertices =
            idToVertex(snapshotSettings.getVerticesToTrigger(), executionGraph);

        // 获取所有需要checkpoint确认的顶点，即所有的顶点
        List<ExecutionJobVertex> ackVertices =
            idToVertex(snapshotSettings.getVerticesToAcknowledge(), executionGraph);

        // 获取所有需要接收到提交checkpoint信息的顶点，即所有的顶点
        List<ExecutionJobVertex> confirmVertices =
            idToVertex(snapshotSettings.getVerticesToConfirm(), executionGraph);

        // Maximum number of remembered checkpoints
        // 获取历史记录checkpoint最大数量
        int historySize = jobManagerConfig.getInteger(WebOptions.CHECKPOINTS_HISTORY_SIZE);

        // 创建checkpoint状态跟踪器，和CheckpointCoordinator配合工作
        CheckpointStatsTracker checkpointStatsTracker =
            new CheckpointStatsTracker(
            historySize,
            ackVertices,
            snapshotSettings.getCheckpointCoordinatorConfiguration(),
            metrics);

        // load the state backend from the application settings
        // 获取状态后端的配置
        final StateBackend applicationConfiguredBackend;
        final SerializedValue<StateBackend> serializedAppConfigured =
            snapshotSettings.getDefaultStateBackend();

        if (serializedAppConfigured == null) {
            applicationConfiguredBackend = null;
        } else {
            try {
                // 根据应用的配置获取状态后端
                applicationConfiguredBackend =
                    serializedAppConfigured.deserializeValue(classLoader);
            } catch (IOException | ClassNotFoundException e) {
                throw new JobExecutionException(
                    jobId, "Could not deserialize application-defined state backend.", e);
            }
        }

        final StateBackend rootBackend;
        try {
            // 获取状态后端配置
            // 如果应用的状态后端没有配置，使用配置文件中的状态后端
            // 如果配置文件中也没有，使用默认值
            rootBackend =
                StateBackendLoader.fromApplicationOrConfigOrDefault(
                applicationConfiguredBackend, jobManagerConfig, classLoader, log);
        } catch (IllegalConfigurationException | IOException | DynamicCodeLoadingException e) {
            throw new JobExecutionException(
                jobId, "Could not instantiate configured state backend", e);
        }
        
        // load the checkpoint storage from the application settings
        // 从app设置中加载checkpoint存储配置
        final CheckpointStorage applicationConfiguredStorage;
        final SerializedValue<CheckpointStorage> serializedAppConfiguredStorage =
            snapshotSettings.getDefaultCheckpointStorage();

        if (serializedAppConfiguredStorage == null) {
            applicationConfiguredStorage = null;
        } else {
            try {
                applicationConfiguredStorage =
                    serializedAppConfiguredStorage.deserializeValue(classLoader);
            } catch (IOException | ClassNotFoundException e) {
                throw new JobExecutionException(
                    jobId,
                    "Could not deserialize application-defined checkpoint storage.",
                    e);
            }
        }
        
        // 和状态后端的配置类似，从应用配置和flink配置文件中加载checkpoint存储配置
        final CheckpointStorage rootStorage;
        try {
            rootStorage =
                CheckpointStorageLoader.load(
                applicationConfiguredStorage,
                null,
                rootBackend,
                jobManagerConfig,
                classLoader,
                log);
        } catch (IllegalConfigurationException | DynamicCodeLoadingException e) {
            throw new JobExecutionException(
                jobId, "Could not instantiate configured checkpoint storage", e);
        }

        // instantiate the user-defined checkpoint hooks

        // 实例化用户定义的checkpoint钩子
        // 这些钩子可以在恢复快照或者是触发快照的时候执行
        final SerializedValue<MasterTriggerRestoreHook.Factory[]> serializedHooks =
            snapshotSettings.getMasterHooks();
        final List<MasterTriggerRestoreHook<?>> hooks;

        if (serializedHooks == null) {
            hooks = Collections.emptyList();
        } else {
            final MasterTriggerRestoreHook.Factory[] hookFactories;
            try {
                hookFactories = serializedHooks.deserializeValue(classLoader);
            } catch (IOException | ClassNotFoundException e) {
                throw new JobExecutionException(
                    jobId, "Could not instantiate user-defined checkpoint hooks", e);
            }

            final Thread thread = Thread.currentThread();
            final ClassLoader originalClassLoader = thread.getContextClassLoader();
            thread.setContextClassLoader(classLoader);

            try {
                hooks = new ArrayList<>(hookFactories.length);
                for (MasterTriggerRestoreHook.Factory factory : hookFactories) {
                    hooks.add(MasterHooks.wrapHook(factory.create(), classLoader));
                }
            } finally {
                thread.setContextClassLoader(originalClassLoader);
            }
        }

        // 获取checkpoint协调器的配置
        final CheckpointCoordinatorConfiguration chkConfig =
            snapshotSettings.getCheckpointCoordinatorConfiguration();

        // 为executionGraph应用checkpoint的相关配置
        executionGraph.enableCheckpointing(
            chkConfig,
            triggerVertices,
            ackVertices,
            confirmVertices,
            hooks,
            checkpointIdCounter,
            completedCheckpointStore,
            rootBackend,
            checkpointStatsTracker,
            checkpointsCleaner);
    }

    // create all the metrics for the Execution Graph
    // 创建相关监控项，监控任务运行时间，重启时间和停止时间
    metrics.gauge(RestartTimeGauge.METRIC_NAME, new RestartTimeGauge(executionGraph));
    metrics.gauge(DownTimeGauge.METRIC_NAME, new DownTimeGauge(executionGraph));
    metrics.gauge(UpTimeGauge.METRIC_NAME, new UpTimeGauge(executionGraph));

    return executionGraph;
}

创建ExecutionGraph的主要步骤大致如下：

• 获取Job信息
• 创建ExecutionGraph本体
• 绑定JobGraph顶点
• 设置Checkpoint配置
• 设置状态后端配置
• 设置Checkpoint存储
• 设置Checkpoint钩子
• 设置作业监控

ExecutionGraph构造函数

• jobInformation：作业信息的一个封装，包含作业id，名称，配置，用户代码和classpath等。
• futureExecutor：异步执行线程池。
• ioExecutor：IO操作线程池。
• rpcTimeout：RPC调用超时时间。
• maxPriorAttemptsHistoryLength：保留在历史记录中的最大重试次数。
• classLoader：用户代码类加载器。
• blobWriter：用于将数据写入blob server。
• partitionReleaseStrategyFactory：IntermediateResultPartition释放策略工厂类。
• shuffleMaster：用于注册IntermediateResultPartition（中间结果分区），向JobMaster注册数据分区及它的生产者。
• partitionTracker：用于追踪和释放分区。
• partitionLocationConstraint：限制在部署的时候partition的位置可否未知。在批模式，分区未知可以未知，但是在流模式，分区位置必须是已知的。
• executionDeploymentListener：执行计划部署监听器
• executionStateUpdateListener：执行计划更新监听器
• initializationTimestamp：初始时间戳
• vertexAttemptNumberStore：用于储存每个Job顶点重试次数

attachJobGraph方法

该方法将JobGraph绑定到ExecutionGraph。

@Override
public void attachJobGraph(List<JobVertex> topologiallySorted) throws JobException {

    // 检查在JobMaster主线程执行
    assertRunningInJobMasterMainThread();

    LOG.debug(
        "Attaching {} topologically sorted vertices to existing job graph with {} "
        + "vertices and {} intermediate results.",
        topologiallySorted.size(),
        tasks.size(),
        intermediateResults.size());

    // 创建保存Execution作业顶点的集合
    final ArrayList<ExecutionJobVertex> newExecJobVertices =
        new ArrayList<>(topologiallySorted.size());
    final long createTimestamp = System.currentTimeMillis();

    for (JobVertex jobVertex : topologiallySorted) {

        // 如果有顶点是数据输入顶点并且是无法停止的顶点
        // 则设置ExecutionGraph的数据源task属性为无法停止
        if (jobVertex.isInputVertex() && !jobVertex.isStoppable()) {
            this.isStoppable = false;
        }

        // create the execution job vertex and attach it to the graph
        // 创建出ExecutionJobVertex
        ExecutionJobVertex ejv =
            new ExecutionJobVertex(
            this,
            jobVertex,
            maxPriorAttemptsHistoryLength,
            rpcTimeout,
            createTimestamp,
            this.initialAttemptCounts.getAttemptCounts(jobVertex.getID()));

        // 设置前一个节点的IntermediateResult给当前ejv
        // 完成连接到前置节点这个逻辑，即这个方法名的含义
        ejv.connectToPredecessors(this.intermediateResults);

        // 将job顶点ID和ejv作为键值对，放入ExecutionGraph
        ExecutionJobVertex previousTask = this.tasks.putIfAbsent(jobVertex.getID(), ejv);
        
        // 如果previousTask不为空，说明两个JobGraph的顶点具有相同的ID，为异常情况
        if (previousTask != null) {
            throw new JobException(
                String.format(
                    "Encountered two job vertices with ID %s : previous=[%s] / new=[%s]",
                    jobVertex.getID(), ejv, previousTask));
        }

        // 遍历ejv中创建的IntermediateResult
        // 该IntermediateResult在ExecutionJobVertex构造函数创建
        // 从ejv对应JobVertex的IntermediateDataSets创建出IntermediateResult
        for (IntermediateResult res : ejv.getProducedDataSets()) {
            IntermediateResult previousDataSet =
                this.intermediateResults.putIfAbsent(res.getId(), res);
            // 同理，检查result的ID不能重复
            if (previousDataSet != null) {
                throw new JobException(
                    String.format(
                        "Encountered two intermediate data set with ID %s : previous=[%s] / new=[%s]",
                        res.getId(), res, previousDataSet));
            }
        }

        // 该集合按照顶点创建顺序保存ejv，将ejv保存起来
        this.verticesInCreationOrder.add(ejv);
        // 统计总的顶点数，作业实际执行的时候，每个并行度都会部署一个vertex运行task
        // 因此需要累加各个ejv的并行度
        this.numVerticesTotal += ejv.getParallelism();
        newExecJobVertices.add(ejv);
    }

    // 注册所有的ExecutionVertex和它输出数据的IntermediateResultPartition
    // ExecutionVertex为物理执行节点，一个ExecutionJobVertex有多少并行度，就会包含多少个ExecutionVertex
    registerExecutionVerticesAndResultPartitions(this.verticesInCreationOrder);

    // the topology assigning should happen before notifying new vertices to failoverStrategy
    // 创建执行拓扑
    //执行拓扑包含所有ExecutionVertex，ResultPartition以及PipelinedRegion
    executionTopology = DefaultExecutionTopology.fromExecutionGraph(this);

    // 创建分区释放策略
    partitionReleaseStrategy =
        partitionReleaseStrategyFactory.createInstance(getSchedulingTopology());
}

Pipelined Region

创建executionTopology我们会遇到Pipelined region。在解释这个概念前需要了解下pipelined result和blocking result的区别。

Pipelined result指的是数据从管道中源源不断的流出，下游可以连续消费产生的数据。一旦上游产生数据，下游就可以立即开始消费。Pipelined result生产数据的过程永远不会停止。此类型对应的作业为流计算作业。

Blocking result只能等到上游数据生产过程结束的时候才可以消费。Blocking result永远是有限的。典型的场景是批处理作业。

Pipelined Region是ExecutionGraph的一部分。它包含连续多个pipeline类型数据交换task（生成pipelined result）。因此一个ExecutionGraph可被分隔为多个pipelined Region，他们之间有block类型作业相连接。

Pipelined Region的意义是region内部的所有消费者必须持续消费上游生产者产生的数据，从而避免阻塞上游，或者产生反压。所以说同一个pipelined Region内的所有task启动或失败之时都必须同时被调度或重启。

Pipelined region调度的官网解释请参见：https://flink.apache.org/2020/12/15/pipelined-region-sheduling.html

enableCheckpointing方法

该方法为ExecutionGraph初始化检查点相关配置。主要逻辑是创建和配置CheckpointCoordinator对象。代码如下所示：

@Override
public void enableCheckpointing(
    CheckpointCoordinatorConfiguration chkConfig,
    List<MasterTriggerRestoreHook<?>> masterHooks,
    CheckpointIDCounter checkpointIDCounter,
    CompletedCheckpointStore checkpointStore,
    StateBackend checkpointStateBackend,
    CheckpointStorage checkpointStorage,
    CheckpointStatsTracker statsTracker,
    CheckpointsCleaner checkpointsCleaner) {

    // 检查作业必须处于已创建状态
    checkState(state == JobStatus.CREATED, "Job must be in CREATED state");
    // 检查CheckpointCoordinator（检查点协调器）必须未创建，避免重复操作
    checkState(checkpointCoordinator == null, "checkpointing already enabled");

    // 收集各个ExecutionJobVertex的OperatorCoordinator
    // OperatorCoordinator运行在JobManager，与作业vertex中的operator相关联。用于和operator交互
    final Collection<OperatorCoordinatorCheckpointContext> operatorCoordinators =
        buildOpCoordinatorCheckpointContexts();

    checkpointStatsTracker = checkNotNull(statsTracker, "CheckpointStatsTracker");

    // 创建checkpoint失败管理器，负责在checkpoint失败时候调用处理逻辑
    CheckpointFailureManager failureManager =
        new CheckpointFailureManager(
        chkConfig.getTolerableCheckpointFailureNumber(),
        new CheckpointFailureManager.FailJobCallback() {
            @Override
            public void failJob(Throwable cause) {
                getJobMasterMainThreadExecutor().execute(() -> failGlobal(cause));
            }

            @Override
            public void failJobDueToTaskFailure(
                Throwable cause, ExecutionAttemptID failingTask) {
                getJobMasterMainThreadExecutor()
                    .execute(
                    () ->
                    failGlobalIfExecutionIsStillRunning(
                        cause, failingTask));
            }
        });

    // 创建CheckpointCoordinator周期自动触发checkpoint的定时器
    checkState(checkpointCoordinatorTimer == null);

    checkpointCoordinatorTimer =
        Executors.newSingleThreadScheduledExecutor(
        new DispatcherThreadFactory(
            Thread.currentThread().getThreadGroup(), "Checkpoint Timer"));

    // 创建CheckpointCoordinator，负责协调整个集群范围内所有operator的checkpoint操作，发起checkpoint操作和提交checkpoint
    // create the coordinator that triggers and commits checkpoints and holds the state
    checkpointCoordinator =
        new CheckpointCoordinator(
        jobInformation.getJobId(),
        chkConfig,
        operatorCoordinators,
        checkpointIDCounter,
        checkpointStore,
        checkpointStorage,
        ioExecutor,
        checkpointsCleaner,
        new ScheduledExecutorServiceAdapter(checkpointCoordinatorTimer),
        SharedStateRegistry.DEFAULT_FACTORY,
        failureManager,
        createCheckpointPlanCalculator(),
        new ExecutionAttemptMappingProvider(getAllExecutionVertices()));

    // register the master hooks on the checkpoint coordinator
    // 设置主消息钩子，在创建checkpoint或从checkpoint恢复的时候回调
    for (MasterTriggerRestoreHook<?> hook : masterHooks) {
        if (!checkpointCoordinator.addMasterHook(hook)) {
            LOG.warn(
                "Trying to register multiple checkpoint hooks with the name: {}",
                hook.getIdentifier());
        }
    }

    // 配置checkpoint状态跟踪器
    checkpointCoordinator.setCheckpointStatsTracker(checkpointStatsTracker);

    // interval of max long value indicates disable periodic checkpoint,
    // the CheckpointActivatorDeactivator should be created only if the interval is not max
    // value
    // 如果没有禁用周期性触发checkpoint，注册一个作业状态监听器
    // 该listener为CheckpointCoordinator所用，监听作业状态的变化
    if (chkConfig.getCheckpointInterval() != Long.MAX_VALUE) {
        // the periodic checkpoint scheduler is activated and deactivated as a result of
        // job status changes (running -> on, all other states -> off)
        registerJobStatusListener(checkpointCoordinator.createActivatorDeactivator());
    }

    // 配置状态后端名称和checkpoint存储名称
    this.stateBackendName = checkpointStateBackend.getClass().getSimpleName();
    this.checkpointStorageName = checkpointStorage.getClass().getSimpleName();
}

从ExecutionGraph到部署Task

ExecutionVertex每次执行都会创建出一个Execution对象。

在JobManager启动完毕之后，会对Scheduler发出开始调度的命令（调用SchedulerBase的startScheduling方法）。经过中间层层调用（较为复杂，这里暂时省略），最终到达Execution.deploy方法。

Execution.deploy方法为真正的部署运行逻辑，根据task资源和ExecutionVertex构造出一个task部署描述符。这个部署描述符的作用为携带task执行配置，通过RPC的方式传递给TaskManager，从而创建出一个符合要求的task。

public void deploy() throws JobException {
    // 确保在JobMaster主线程执行
    assertRunningInJobMasterMainThread();

    // 获取分配的资源
    final LogicalSlot slot = assignedResource;

    checkNotNull(
        slot,
        "In order to deploy the execution we first have to assign a resource via tryAssignResource.");

    // Check if the TaskManager died in the meantime
    // This only speeds up the response to TaskManagers failing concurrently to deployments.
    // The more general check is the rpcTimeout of the deployment call
    if (!slot.isAlive()) {
        throw new JobException("Target slot (TaskManager) for deployment is no longer alive.");
    }

    // make sure exactly one deployment call happens from the correct state
    // note: the transition from CREATED to DEPLOYING is for testing purposes only
    // 获取之前的状态，并切换状态为正在部署（DEPLOYING）
    ExecutionState previous = this.state;
    if (previous == SCHEDULED || previous == CREATED) {
        if (!transitionState(previous, DEPLOYING)) {
            // race condition, someone else beat us to the deploying call.
            // this should actually not happen and indicates a race somewhere else
            throw new IllegalStateException(
                "Cannot deploy task: Concurrent deployment call race.");
        }
    } else {
        // vertex may have been cancelled, or it was already scheduled
        throw new IllegalStateException(
            "The vertex must be in CREATED or SCHEDULED state to be deployed. Found state "
            + previous);
    }

    // 检查这个slot资源是否分配给了当前execution
    if (this != slot.getPayload()) {
        throw new IllegalStateException(
            String.format(
                "The execution %s has not been assigned to the assigned slot.", this));
    }

    try {

        // race double check, did we fail/cancel and do we need to release the slot?
        // 再次检查作业状态是否为正在部署
        if (this.state != DEPLOYING) {
            slot.releaseSlot(
                new FlinkException(
                    "Actual state of execution "
                    + this
                    + " ("
                    + state
                    + ") does not match expected state DEPLOYING."));
            return;
        }

        LOG.info(
            "Deploying {} (attempt #{}) with attempt id {} to {} with allocation id {}",
            vertex.getTaskNameWithSubtaskIndex(),
            attemptNumber,
            vertex.getCurrentExecutionAttempt().getAttemptId(),
            getAssignedResourceLocation(),
            slot.getAllocationId());

        // 创建一个Task部署描述符
        // 该部署描述符携带了ExecutionVertex及其分配的资源等信息
        final TaskDeploymentDescriptor deployment =
            TaskDeploymentDescriptorFactory.fromExecutionVertex(vertex, attemptNumber)
            .createDeploymentDescriptor(
            slot.getAllocationId(),
            taskRestore,
            producedPartitions.values());

        // null taskRestore to let it be GC'ed
        taskRestore = null;

        final TaskManagerGateway taskManagerGateway = slot.getTaskManagerGateway();

        final ComponentMainThreadExecutor jobMasterMainThreadExecutor =
            vertex.getExecutionGraphAccessor().getJobMasterMainThreadExecutor();

        getVertex().notifyPendingDeployment(this);
        // We run the submission in the future executor so that the serialization of large TDDs
        // does not block
        // the main thread and sync back to the main thread once submission is completed.
        // 在这里，异步调用taskManagerGateway，通过rpc方式通知TaskManager
        // 将Task部署描述符发送给TaskManager
        // TaskManager接收到后开始部署Task
        CompletableFuture.supplyAsync(
            () -> taskManagerGateway.submitTask(deployment, rpcTimeout), executor)
            .thenCompose(Function.identity())
            .whenCompleteAsync(
            (ack, failure) -> {
                if (failure == null) {
                    vertex.notifyCompletedDeployment(this);
                } else {
                    if (failure instanceof TimeoutException) {
                        String taskname =
                            vertex.getTaskNameWithSubtaskIndex()
                            + " ("
                            + attemptId
                            + ')';

                        markFailed(
                            new Exception(
                                "Cannot deploy task "
                                + taskname
                                + " - TaskManager ("
                                + getAssignedResourceLocation()
                                + ") not responding after a rpcTimeout of "
                                + rpcTimeout,
                                failure));
                    } else {
                        markFailed(failure);
                    }
                }
            },
            jobMasterMainThreadExecutor);

    } catch (Throwable t) {
        markFailed(t);
    }
}

上面方法中通过TaskManagerGateway调用了TaskExecutor的submitTask方法。

@Override
public CompletableFuture<Acknowledge> submitTask(
    TaskDeploymentDescriptor tdd, JobMasterId jobMasterId, Time timeout) {
    // ...
    // 创建出Task
    Task task =
                    new Task(
                            jobInformation,
                            taskInformation,
                            tdd.getExecutionAttemptId(),
                            tdd.getAllocationId(),
                            tdd.getSubtaskIndex(),
                            tdd.getAttemptNumber(),
                            tdd.getProducedPartitions(),
                            tdd.getInputGates(),
                            memoryManager,
                            taskExecutorServices.getIOManager(),
                            taskExecutorServices.getShuffleEnvironment(),
                            taskExecutorServices.getKvStateService(),
                            taskExecutorServices.getBroadcastVariableManager(),
                            taskExecutorServices.getTaskEventDispatcher(),
                            externalResourceInfoProvider,
                            taskStateManager,
                            taskManagerActions,
                            inputSplitProvider,
                            checkpointResponder,
                            taskOperatorEventGateway,
                            aggregateManager,
                            classLoaderHandle,
                            fileCache,
                            taskManagerConfiguration,
                            taskMetricGroup,
                            resultPartitionConsumableNotifier,
                            partitionStateChecker,
                            getRpcService().getExecutor());
// ...
}

到这里为止，TaskManager中的具体任务Task对象已经被创建出来。从JobGraph生成ExecutionGraph并最终部署为Task的过程已分析完毕。

本博客为作者原创，欢迎大家参与讨论和批评指正。如需转载请注明出处。