引言

前面我們?cè)诮榻BDataStream時(shí)，介紹了Flink任務(wù)提交時(shí)從StreamGraph->JobGraph->ExecutionGraph的過(guò)程，而如何生成ExecutionGraph并沒(méi)介紹，本節(jié)來(lái)介紹在具體調(diào)度執(zhí)行時(shí)使用的圖結(jié)構(gòu)ExecutionGraph。StreamGraph和JobGraph是在Client生成的。

ExecutionGraph是在JobManager(Flink任務(wù)執(zhí)行時(shí)的Master節(jié)點(diǎn))端生成的，JobManager會(huì)根據(jù)提交的JobGraph來(lái)生成ExecutionGraph。

重要類(lèi)

DefaultExecutionGraph

ExecutionGraph的實(shí)現(xiàn)類(lèi)，保存了具體的Graph結(jié)構(gòu)信息、具體執(zhí)行時(shí)的作業(yè)和任務(wù)相關(guān)信息以及作業(yè)執(zhí)行中的中間結(jié)果信息等。相關(guān)重要屬性如下

    // JobGraph的節(jié)點(diǎn)ID和ExecutionGraph的節(jié)點(diǎn)信息映射
    private final Map<JobVertexID, ExecutionJobVertex> tasks;
    // 按依賴(lài)順序的Execution節(jié)點(diǎn)數(shù)據(jù)
    private final List<ExecutionJobVertex> verticesInCreationOrder;
    // 執(zhí)行嘗試信息
    private final Map<ExecutionAttemptID, Execution> currentExecutions;
    //中間結(jié)果數(shù)據(jù)信息
    private final Map<IntermediateDataSetID, IntermediateResult> intermediateResults;
    // 當(dāng)前作業(yè)狀態(tài)
    private volatile JobStatus state = JobStatus.CREATED;
    //執(zhí)行拓?fù)浣Y(jié)構(gòu)
    private DefaultExecutionTopology executionTopology;
    // checkpoint處理協(xié)調(diào)器
    @Nullable private CheckpointCoordinator checkpointCoordinator;

ExecutionJobVertex

在ExecutionGraph中的節(jié)點(diǎn)信息，與JobGraph的JobVertex是一一對(duì)應(yīng)的。其中存儲(chǔ)了

    // 每個(gè)子任務(wù)節(jié)點(diǎn)信息
    @Nullable private ExecutionVertex[] taskVertices;
    // 產(chǎn)出數(shù)據(jù)集
    @Nullable private IntermediateResult[] producedDataSets;
    // 輸入數(shù)據(jù)集
    @Nullable private List<IntermediateResult> inputs;
    // 并行度
    private final VertexParallelismInformation parallelismInfo;

ExecutionVertex

ExecutionJobVertex中根據(jù)并行度生成的單個(gè)子任務(wù),包括具體的子任務(wù)的編號(hào)，執(zhí)行信息等

IntermediateResult

節(jié)點(diǎn)的每個(gè)輸出鏈對(duì)應(yīng)一個(gè)IntermediateResult，每個(gè)IntermediateResult下按ExecutionJobVertex的并行度對(duì)應(yīng)有相應(yīng)的IntermediateResultPartition。

SlotSharingGroup

定義不同節(jié)點(diǎn)的任務(wù)可以部署到同一個(gè)slot中，對(duì)slot進(jìn)行共享，更為有效的使用slot資源。

ExecutionGraph生成

ExecutionGraph是Scheduler(JobManager中的負(fù)責(zé)調(diào)度處理的類(lèi))中實(shí)例化時(shí)通過(guò)調(diào)用createAndRestoreExecutionGraph方法來(lái)生成ExecutionGraph的。其最終調(diào)用的是DefaultExecutionGraphBuilder類(lèi)中的buildGraph()方法。其具體流程如下：

• 創(chuàng)建一個(gè)DefaultExecutionGraph的實(shí)例，這里主要是傳入一些參數(shù)處理，并沒(méi)有關(guān)聯(lián)JobGraph的信息；
• 初始化JobVertex,處理inputoutput的格式信息；
• 將JobGraph的所有JobVertex進(jìn)行按依賴(lài)順序進(jìn)行排序處理；
• 調(diào)用ExecutionGraph的attachJobGraph方法將JobVertex列表信息綁定到ExecutionGraph
  每一個(gè)ExecutionJobVertex對(duì)應(yīng)一個(gè)JobVertex，每一個(gè)IntermediateResult對(duì)應(yīng)到一個(gè)JobVertex的IntermediateDataSet，再根據(jù)JobVertex的并行度生成對(duì)應(yīng)數(shù)量的ExecutionVertex,用數(shù)組存儲(chǔ)
  根據(jù)JobVertex的inputs信息初始化ExecutionJobVertex的inputs信息。
• 配置statebackend和checkpoint信息，此部分留到介紹checkpoint時(shí)再詳細(xì)介紹

總結(jié)

本篇接著01-DataStream基礎(chǔ)介紹了JobGraph到ExecutionGraph的轉(zhuǎn)換過(guò)程。首先介紹了ExecutionGraph中的相關(guān)核心概念，如ExecutionJobVertex、IntermediateResult等。后面介紹了ExecutionGraph的詳細(xì)生成過(guò)程。

在ExecutionGraph生成的最后會(huì)設(shè)置checkpoint等信息，此塊后面單獨(dú)介紹。ExecutionGraph生成好后，會(huì)通過(guò)DefaultScheduler的startScheduling()方法來(lái)觸發(fā)進(jìn)行調(diào)度(具體調(diào)度及運(yùn)行后面介紹)，更多關(guān)于Flink ExecutionGraph生成的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

最新評(píng)論