可以關注下公眾號，專注資料開發、資料架構之路，熱衷于分享技術干貨，

一、提交流程

Flink作業在開發完畢之后，需要提交到Flink集群執行，ClientFronted是入口，觸發用戶開發的Flink應用Jar檔案中的main方法，然后交給PipelineExecutor（流水線執行器，在FlinkClient 升成JobGraph之后，將作業提交給集群的重要環節，）#execue方法，最侄訓選擇一個觸發一個具體的PiplineExecutor執行，

提交模式又可分為：

• Detached：Flink Client創建完集群之后，可以退出命令列視窗，集群獨立運行，
• Attached：不能關閉命令列視窗，需要與集群之間維持連接，

1.1 Yarn Session提交流程

啟動集群：

• 使用bin/yarn-session.sh提交會話模式的作業，

如果提交到已經存在的集群，則獲取Yarn集群資訊、應用ID，并準備提交作業，如果啟動新的Yarn Session集群，則進入步驟（2）

• Yarn啟動新Flink集群

1）如果沒有集群，則創建一個新的Session模式的集群，首先將應用配置（flink-conf.yaml、logback.xml、log4j.properties）和相關檔案（Flink Jar、配置類檔案、用戶Jar檔案、JobGraph物件等）上傳至分布式存盤(如HDFS)的應用暫存目錄，

2）通過Yarn Client 向Yarn 提交Flink創建集群的申請，Yarn分配資源，在申請的Yarn Container中初始化并啟動FlinkJobManager行程，在JobManager行程中運行YarnSessionClusterEntrypoint作為集群啟動入口（不同的集群部署模式有不同的ClusterEntrypoint實作），初始化Dispatcher、ResourceManager，啟動相關的RPC服務，等待Client通過Rest介面提交作業，

作業提交：

Yarn 集群準備好后，開始作業提交，

1）Flink Client通過Rest向Dispatcher提交JobGraph，

2）Dispatcher是Rest介面，不負責實際的調度、執行方面的作業，當收到JobGraph后，為作業創建一個JobMaster，將作業交給JobManager（負責作業調度、管理作業和Task的生命周期），構建ExecutionGraph（JobGraph的并行化版本，調度層最核心的資料結構），

這兩個步驟結束后，作業進入調度執行階段，

作業調度執行：

1）JobMaster向YarnResourceManager申請資源，開始調度ExecutionGraph執行，向YarnResourceManager申請資源；初次提交作業集群中尚沒有TaskManager，此時資源不足，開始申請資源，

2）YarnResourceManager收到JobManager的資源請求，如果當前有空閑Slot則將Slot分配給JobMaster.，否則YarnResourceManager將向YarnMaster請求創建TaskManager，

3）YarnResourceManager將資源請求加入到等待請求佇列，并通過心跳向Yarn RM 申請新的Container資源來啟動TaskManager行程，Yarn分配新的Container給TaskManager，

4）YarnResourceManager啟動，然后從HDFS加載Jar檔案等所需要的的相關資源，在容器中啟動TaskManager，

5）TaskManager啟動之后，向ResourceManager注冊，并把自己的Slot資源情況匯報給ResouceManager，

6）ResourceManager從等待佇列中取出Slot請求，向TaskManager確認資源可用情況，并告知TaskManager將Slot分配給哪個JobMaster，

7）TaskManager向JobMaster提供Slot，JobMaster調度Task到TaskManager的此Slot上執行，

1.2 Yarn Per-Job提交流程

啟動集群：

• 使用./flink run -m yarn-cluster提交Per-Job模式的作業，
• Yarn啟動Flink集群，該模式下Flink集群的啟動入口是YarnJobClusterEntryPoint，其他與YarnSession模式下集群的啟動類似，

作業提交：

該步驟與Seesion模式下的不同，Client并不會通過Rest向Dispatcher提交JobGraph,由Dispatcher從本地檔案系統獲取JObGraph，其后的不好走與Session模式的一樣

作業調度執行：

與Yarn Session模式下一致，

1.3 K8s Session提交流程

啟動集群：

• Flink客戶端首先連接Kubernetes API Server，提交Flink集群的資源描述檔案，包括flink-configuration-configmap.yaml、jobmanager-service.yaml、jobmanager-deployment.yaml和taskmanager-deployment.yaml等，
• Kubernets Master會根據這些資源描述檔案去創建對應的Kubernetes物體，以JobManager部署為例，Kubernetes集群中的某個節點收到請求后，Kubelet行程會從中央倉庫下載Flink鏡像，準備和掛載卷，然后執行啟動命令，Pod啟動后Flink Master(JobManager)行程隨之啟動，初始化Dispacher和KubernetesResourceManager，并通過K8s服務對外暴露FlinkMaster的埠，K8s服務類似于路由服務，

兩個步驟完成之后，Session模式的集群就創建成功，集群可以接收作業提交請求，但是此時還沒有JobManager、TaskManager，當作業需要執行時，才會按需創建，

作業提交：

• Client用戶可以通過Flink命令列（即Flink Client）向這個會話模式的集群提交任務，此時JobGraph會在FlinkClient端生成，然后和用戶Jar包一起通過RestClient上傳，
• 作業提交成功，Dispatcher會為每個作業啟動一個JobMaster，將JobGraph交給JobMaster調度執行，

兩個步驟完成之后，作業進入調度執行階段，

作業調度執行：

K8s Session模式集群下，ResourceManager向k8sMaster申請和釋放TaskManager，除此之外，作業的調度與執行和Yarn模式是一樣的，

1）JobMaster向KubernetesResourceManager請求Slot，

2）KubernetesResourceManager從kubernetes集群分配TaskManager，每個TaskManager都是具有唯一標識的Pod，KubernetesResourceManager會為TaskManager生成一份新的組態檔，里面有Flink Master的service name 作為地址，這樣在FLInkMaster failover之后，TaskManager仍然可以重新連上，

3）Kubernetes集群分配一個新的Pod后，在上面啟動TaskManager，

4）TaskManager啟動后注冊到SlotManager，

5）SlotManager向TaskManager請求Slot.

6）TaskManager 提供Slot給JobMaster，然后任務就會被分配到這個Slot上運行，

二、Graph總覽

• 流計算應用的Graph轉換：StreamGraph-->JobGraph-->ExecutionGraph-->物理執行圖(啟動計算任務)
• 批處理應用的Graph轉換：OptimizedPlan-->JobGraph
• Table & SQL API的Graph轉換：Blink Table Planner /Flink Table Planner，

2.1 流圖

使用DataStreamAPI 開發的應用程式，首先被轉換為Transformation,然后被映射為StreamGraph，

2.1.1 SteramGraph核心物件

• StreamNode

StreamNode是StremGraph中的節點 ，從Transformation轉換而來，可以簡單理解為一個StreamNode表示一個算子，從邏輯上來說，SteramNode在StreamGraph中存在物體和虛擬的StreamNode，StremNode可以有多個輸入，也可以有多個輸出，

物體的StreamNode會最終變成物理算子，虛擬的StreamNode會附著在StreamEdge上，

• StreamEdge

StreamEdge是StreamGraph中的邊，用來連接兩個StreamNode,一個StreamNode可以有多個出邊、入邊，StreamEdge中包含了旁路輸出、磁區器、欄位篩選輸出等資訊，

2.1.2 StreamGraph生成程序

StreamGraph在FlinkClient中生成，由FlinkClient在提交的時候觸發Flink應用的main方法，用戶撰寫的業務邏輯組裝成Transformation流水線，在最后呼叫StreamExecutionEnvironment.execute() 的時候開始觸發StreamGraph構建，

StreamGraph實際上是在StreamGraphGenerator中生成的，從SinkTransformation（輸出） 向前追溯到SourceTransformation，在遍歷程序中一邊遍歷一邊構建StreamGraph，

在遍歷Transformation的程序中，會對不同型別的Transformation分別進行轉換，對于物理Transformation則轉換為StreamNode物體，對于虛擬Transformation則作為虛擬StreamNode，

針對于某一種型別的Transformation，會呼叫其相應的transformxxx()函式進行轉換，transfromxxx（）首先轉換上游Transformation進行遞回轉換，確保上游的都已經完成了轉換，然后通過addOperator()方法構造出StreamNode，通過addEdge()方法與上游的transform進行連接，構造出StreamEdge，

在添加StreamEdge的程序中，如果ShuffleMode為null，則使用ShuffleMode PIPELINED模式，在流計算中，只有PIPLINED模式才會在批處理中設計其他模式，構建StreamEdge的時候，在轉換Transformation程序中生成的 虛擬StreamNode會將虛擬StreamNode的資訊附著在StreamEdge上

2.1.3 虛擬Transformation 的轉換

虛擬的Transformation生成的時候不會轉換為SteramNode，而是添加為虛擬節點，

	private void addEdgeInternal(Integer upStreamVertexID,
			Integer downStreamVertexID,
			int typeNumber,
			StreamPartitioner<?> partitioner,
			List<String> outputNames,
			OutputTag outputTag,
			ShuffleMode shuffleMode) {
		//當上游是sideoutput時，遞回呼叫，并傳入sideoutput資訊
		if (virtualSideOutputNodes.containsKey(upStreamVertexID)) {
			int virtualId = upStreamVertexID;
			upStreamVertexID = virtualSideOutputNodes.get(virtualId).f0;
			if (outputTag == null) {
				outputTag = virtualSideOutputNodes.get(virtualId).f1;
			}
			addEdgeInternal(upStreamVertexID, downStreamVertexID, typeNumber, partitioner, null, outputTag, shuffleMode);
			
		}
		//當上游是select時，遞回呼叫，并傳入select資訊
		else if (virtualSelectNodes.containsKey(upStreamVertexID)) {
			int virtualId = upStreamVertexID;
			upStreamVertexID = virtualSelectNodes.get(virtualId).f0;
			if (outputNames.isEmpty()) {
				// selections that happen downstream override earlier selections
				outputNames = virtualSelectNodes.get(virtualId).f1;
			}
			addEdgeInternal(upStreamVertexID, downStreamVertexID, typeNumber, partitioner, outputNames, outputTag, shuffleMode);
		}
		//當上游是Partition時，遞回呼叫，并傳入Partition資訊
		else if (virtualPartitionNodes.containsKey(upStreamVertexID)) {
			int virtualId = upStreamVertexID;
			upStreamVertexID = virtualPartitionNodes.get(virtualId).f0;
			if (partitioner == null) {
				partitioner = virtualPartitionNodes.get(virtualId).f1;
			}
			shuffleMode = virtualPartitionNodes.get(virtualId).f2;
			addEdgeInternal(upStreamVertexID, downStreamVertexID, typeNumber, partitioner, outputNames, outputTag, shuffleMode);
		}
		//不是以上邏輯轉換的情況，真正構建StreamEdge
		else {
			StreamNode upstreamNode = getStreamNode(upStreamVertexID);
			StreamNode downstreamNode = getStreamNode(downStreamVertexID);

			// If no partitioner was specified and the parallelism of upstream and downstream
			// operator matches use forward partitioning, use rebalance otherwise.
			//沒有指定partitioner時，會為其選擇forward或者rebalance
			if (partitioner == null && upstreamNode.getParallelism() == downstreamNode.getParallelism()) {
				partitioner = new ForwardPartitioner<Object>();
			} else if (partitioner == null) {
				partitioner = new RebalancePartitioner<Object>();
			}

			if (partitioner instanceof ForwardPartitioner) {
				if (upstreamNode.getParallelism() != downstreamNode.getParallelism()) {
					throw new UnsupportedOperationException("Forward partitioning does not allow " +
							"change of parallelism. Upstream operation: " + upstreamNode + " parallelism: " + upstreamNode.getParallelism() +
							", downstream operation: " + downstreamNode + " parallelism: " + downstreamNode.getParallelism() +
							" You must use another partitioning strategy, such as broadcast, rebalance, shuffle or global.");
				}
			}

			if (shuffleMode == null) {
				shuffleMode = ShuffleMode.UNDEFINED;
			}
			//創建StreamEdge,并將該SteramEdge添加到上游的輸出，下游的輸入，
			StreamEdge edge = new StreamEdge(upstreamNode, downstreamNode, typeNumber, outputNames, partitioner, outputTag, shuffleMode);

			getStreamNode(edge.getSourceId()).addOutEdge(edge);
			getStreamNode(edge.getTargetId()).addInEdge(edge);
		}
	}

2.2 作業圖

JobGraph可以由流計算的StreamGraph和批處理的OptimizedPlan轉換而來，流計算中，在StreamGraph的基礎上進行了一些優化，如果通過OperatorChain機制將算子合并起來，在執行時，調度在同一個Task執行緒上，避免資料的跨執行緒、跨網段的傳遞，

2.2.1 JobGraph核心物件

• JobVertex

經過算子融合優化后符合條件的多個SteramNode可能會融合在一起生成一個JobVertex，即一個JobVertex包含一個或多個算子，JobVertex的輸入是JobEdge，輸出是IntermediateDataSet，

• JobEdge

JobEdge是JobGraph中連接IntermediateDataSet和JobVertex的邊，表示JobGraph中的一個資料流轉通道，其上游資料源是IntermediateDataSet，下游消費者是JobVertex，資料通過JobEdge 由IntermediateDataSet傳遞給JobVertex，

• IntermediateDataSet

中間資料集IntermediateDataSet是一種邏輯結構，用來表示JobVertex的輸出，即該JobVertex中包含的算子會產生的資料集，不同的執行模式下，其對應的結果磁區型別不同，決定了在執行時刻資料交換的模式，

IntermediateDataSet的個數與該JobVertex對應的StreamNode的出邊數量相同，可以是一個或者多個，

2.2.2 JobGraph生成程序

StreamingJobGraphGenerator負責流計算JobGraph的生成，在轉換前需要進行一系列的預處理，

	private JobGraph createJobGraph() {
		preValidate();

		// make sure that all vertices start immediately
		//設定調度模式
		jobGraph.setScheduleMode(streamGraph.getScheduleMode());

		// Generate deterministic hashes for the nodes in order to identify them across
		// submission iff they didn't change.
		//為每個節點生成確定的hashid作為唯一表示，在提交和執行程序中保持不變，
		Map<Integer, byte[]> hashes = defaultStreamGraphHasher.traverseStreamGraphAndGenerateHashes(streamGraph);

		// Generate legacy version hashes for backwards compatibility
		//為了向后保持兼容，為每個節點生成老版本的hash id
		List<Map<Integer, byte[]>> legacyHashes = new ArrayList<>(legacyStreamGraphHashers.size());
		for (StreamGraphHasher hasher : legacyStreamGraphHashers) {
			legacyHashes.add(hasher.traverseStreamGraphAndGenerateHashes(streamGraph));
		}

		Map<Integer, List<Tuple2<byte[], byte[]>>> chainedOperatorHashes = new HashMap<>();
		//真正對SteramGraph進行轉換，生成JobGraph圖
		setChaining(hashes, legacyHashes, chainedOperatorHashes);
		setPhysicalEdges();

		//設定共享slotgroup
		setSlotSharingAndCoLocation();

		setManagedMemoryFraction(
			Collections.unmodifiableMap(jobVertices),
			Collections.unmodifiableMap(vertexConfigs),
			Collections.unmodifiableMap(chainedConfigs),
			id -> streamGraph.getStreamNode(id).getMinResources(),
			id -> streamGraph.getStreamNode(id).getManagedMemoryWeight());
		//配置checkpoint
		configureCheckpointing();

		jobGraph.setSavepointRestoreSettings(streamGraph.getSavepointRestoreSettings());
		//如果有之前的快取檔案的配置，則重新讀入
		JobGraphGenerator.addUserArtifactEntries(streamGraph.getUserArtifacts(), jobGraph);

		// set the ExecutionConfig last when it has been finalized
		try {
			//設定執行環境配置
			jobGraph.setExecutionConfig(streamGraph.getExecutionConfig());
		}
		catch (IOException e) {
			throw new IllegalConfigurationException("Could not serialize the ExecutionConfig." +
					"This indicates that non-serializable types (like custom serializers) were registered");
		}

		return jobGraph;
	}

預處理完畢后，開始構建JobGraph中的點和邊，從Source向下遍歷StreamGraph，逐步創建JObGraph，在創建的程序中同事完成算子融合(OperatorChain)優化，

執行具體的Chain和JobVertex生成、JobEdge的關聯、IntermediateDataSet，從StreamGraph讀取資料的StreamNode開始，遞回遍歷同時將StreamOperator連接在一起，

整理構建的邏輯如下（看上圖！！！）：

1）從Source開始，Source與下游的FlatMap不可連接，Source是起始節點，自己成為一個JobVertx，

2）此時開始一個新的連接分析，FlatMap是起始節點，與下游的KeyedAgg也不可以連接，那么FlatMap自己成為一個JobVertex，

3）此時開始一個新的連接分析，KeyedAgg是起始節點，并且與下游的Sink可以連接，那么遞回地分析Sink節點，構造Sink與其下游是否可以連接，因為Slink沒有下游，所以KeyedAgg和Sink節點連接在一起，共同構成了一個JobVertex，在這個JobVertex中，KeyedAgg是起始節點，index編號為0，sink節點index編號為1.

構建JobVertex的時候需要將StreamNode中的重要配置資訊復制到JobVertex中，構建好JobVertex之后，需要構建JobEdge將JobVertex連接起來，KeyedAgg和Sink之間構成了一個算子連接，連接內部的算子之間無序構成JobEdge進行連接，

在構建JobEdge的時候，很重要的一點是確定上游JobVertex和下游JobVertex的資料交換方式，此時根據ShuffleMode來確定ResultPartition型別，用FlinkPartition來確定JobVertex的連接方式，

Shuffle確定了ResultPartition,那么就可以確定上游JobVertex輸出的IntermediateDataSet的型別了，也就知道JobEdge的輸入IntermediateDataSet，

ForwardPartitioner和RescalePartitioner兩種型別的Partitioner轉換為DistributionPattern.POINTWISE 的分發模式，其他型別的Partitioner統一轉換為DistributionPattern.ALL_TO_ALL模式，

JobGraph的構建和OperatorChain優化：

private List<StreamEdge> createChain(
			Integer startNodeId,
			Integer currentNodeId,
			Map<Integer, byte[]> hashes,
			List<Map<Integer, byte[]>> legacyHashes,
			int chainIndex,
			Map<Integer, List<Tuple2<byte[], byte[]>>> chainedOperatorHashes) {

		if (!builtVertices.contains(startNodeId)) {

			List<StreamEdge> transitiveOutEdges = new ArrayList<StreamEdge>();

			List<StreamEdge> chainableOutputs = new ArrayList<StreamEdge>();
			List<StreamEdge> nonChainableOutputs = new ArrayList<StreamEdge>();

			StreamNode currentNode = streamGraph.getStreamNode(currentNodeId);
			//獲取當前節點的出邊，判斷是否符合OperatorChain的條件
			//分為兩類：chainableoutputs,nonchainableoutputs
			for (StreamEdge outEdge : currentNode.getOutEdges()) {
				if (isChainable(outEdge, streamGraph)) {
					chainableOutputs.add(outEdge);
				} else {
					nonChainableOutputs.add(outEdge);
				}
			}
			//對于chainable的邊，遞回呼叫createchain
			//回傳值添加到transitiveOutEdges中
			for (StreamEdge chainable : chainableOutputs) {
				transitiveOutEdges.addAll(
						createChain(startNodeId, chainable.getTargetId(), hashes, legacyHashes, chainIndex + 1, chainedOperatorHashes));
			}

			//對于無法chain在一起的邊，邊的下游節點作為Operatorchain的Head節點
			//進行遞回呼叫，回傳值添加到transitiveOutEdges中
			for (StreamEdge nonChainable : nonChainableOutputs) {
				transitiveOutEdges.add(nonChainable);
				createChain(nonChainable.getTargetId(), nonChainable.getTargetId(), hashes, legacyHashes, 0, chainedOperatorHashes);
			}

			List<Tuple2<byte[], byte[]>> operatorHashes =
				chainedOperatorHashes.computeIfAbsent(startNodeId, k -> new ArrayList<>());

			byte[] primaryHashBytes = hashes.get(currentNodeId);
			OperatorID currentOperatorId = new OperatorID(primaryHashBytes);

			for (Map<Integer, byte[]> legacyHash : legacyHashes) {
				operatorHashes.add(new Tuple2<>(primaryHashBytes, legacyHash.get(currentNodeId)));
			}

			chainedNames.put(currentNodeId, createChainedName(currentNodeId, chainableOutputs));
			chainedMinResources.put(currentNodeId, createChainedMinResources(currentNodeId, chainableOutputs));
			chainedPreferredResources.put(currentNodeId, createChainedPreferredResources(currentNodeId, chainableOutputs));

			if (currentNode.getInputFormat() != null) {
				getOrCreateFormatContainer(startNodeId).addInputFormat(currentOperatorId, currentNode.getInputFormat());
			}

			if (currentNode.getOutputFormat() != null) {
				getOrCreateFormatContainer(startNodeId).addOutputFormat(currentOperatorId, currentNode.getOutputFormat());
			}
			//如果當前節點是起始節點，則直接創建JobVertex，否則回傳一個空的StreamConfig
			StreamConfig config = currentNodeId.equals(startNodeId)
					? createJobVertex(startNodeId, hashes, legacyHashes, chainedOperatorHashes)
					: new StreamConfig(new Configuration());

			//將StreamNode中的配置資訊序列化到Streamconfig中，
			setVertexConfig(currentNodeId, config, chainableOutputs, nonChainableOutputs);
			//再次判斷，如果是Chain的起始節點，執行connect（）方法，創建JobEdge和IntermediateDataset
			//否則將當前節點的StreamConfig 添加到chainedConfig中，
			if (currentNodeId.equals(startNodeId)) {

				config.setChainStart();
				config.setChainIndex(0);
				config.setOperatorName(streamGraph.getStreamNode(currentNodeId).getOperatorName());

				for (StreamEdge edge : transitiveOutEdges) {
					connect(startNodeId, edge);
				}

				config.setOutEdgesInOrder(transitiveOutEdges);
				config.setTransitiveChainedTaskConfigs(chainedConfigs.get(startNodeId));

			} else {
				chainedConfigs.computeIfAbsent(startNodeId, k -> new HashMap<Integer, StreamConfig>());

				config.setChainIndex(chainIndex);
				StreamNode node = streamGraph.getStreamNode(currentNodeId);
				config.setOperatorName(node.getOperatorName());
				chainedConfigs.get(startNodeId).put(currentNodeId, config);
			}

			config.setOperatorID(currentOperatorId);

			if (chainableOutputs.isEmpty()) {
				config.setChainEnd();
			}
			return transitiveOutEdges;

		} else {
			return new ArrayList<>();
		}
	}

2.2.3 算子融合

一個Operatorchain在同一個Task執行緒內執行，OperatorChain內的算子之間，在同一個執行緒內通過方法呼叫的方式傳遞資料，能減少執行緒之間的切換，減少訊息的序列化/反序列化，無序借助記憶體快取區，也無須通過網路在算子間傳遞資料，可在減少延遲的同時提高整體吞吐量

operatorchain的條件:

1）下游節點的入度為1

2）SteramEdge的下游節點對應的算子不為null

3）StreamEdge的上游節點對應的算子不為null

4）StreamEdge的上下游節點擁有相同的slotSharingGroup，默認都是default.

5）下游算子的連接策略為ALWAYS.

6）上游算子的連接策略為ALWAYS 或者HEAD.

7）StreamEdge的磁區型別為ForwardPartitioner

8）上下游節點的并行度一致

9）當前StreamGraph允許chain

2.3 執行圖

2.3.1 ExecutionGraph核心物件

• ExecutionJobVertex

該物件和JobGraph中的JobVertex一一對應，該物件還包含了一組ExecutionVertex，數量與該JobVertex中所包含的SteramNode的并行度一致，

ExecutionJobVertex用來將一個JobVertex封裝成一ExecutionJobVertex，并以此創建ExecutionVertex、Execution、IntermediateResult和IntermediateResultPartition，用于豐富ExecutionGraph，

在ExecutionJobVertex的建構式中，首先是依據對應的JobVertex的并發度，生成對應個數的ExecutionVertex，其中，一個ExecutionVertex代表一個ExecutionJobVertex的并發子Task，然后是將原來JobVertex的中間結果IntermediateDataSet轉化為ExecutionGrap中IntermediateResult

• ExecutionVertex

ExecutionJobVertex中會對作業進行并行化處理，構造可以并行執行的實體，每個并行執行的實體就是ExecutionVertex.

構建ExecutionVertex的同時，也回構建ExecutionVertex的輸出IntermediateResult，并且將ExecutionEdge輸出為IntermediatePartition，

ExecutionVertex的建構式中，首先會創建IntermediatePartition，并通過IntermediateResult.setPartition（）建立IntermediateResult和IntermediateResultPartition之間的關系，然后生成Execution，并配置資源相關，

• IntermediateResult

IntermediateResult又叫做中間結果集，該物件是個邏輯概念，表示ExecutionJobVertex的輸出，和JobGraph中的IntermediateDataSet一一對應，同樣，一個ExecutionJobVertex可以有多個中間二級果，取決于當前JobVertex有幾個出邊，

一個中間結果集包含多個中間結果磁區IntermediateResultPartition，其個數等于該JobVertex的并發度，

• IntermediateResultPartition

IntermediateResultPartition又叫做中間結果磁區，表示1個ExecutionVertex輸出結果，與ExecutionEdge相關聯，

• ExecutionEdge

表示ExecutionVertex的輸入，連接到上游產生的IntermediateResultPartition，一個Execution對應于唯一的一個IntermediateResultPartition和一個ExecutionVertex，一個ExecutionVertex可以有多個ExecutionEdge，

• Execution

ExecutionVertex相當于每個Task的模板，在真正執行的時候，會將ExecutionVertex中的資訊包裝為一個Execution,執行一個ExecutionVertex的一次嘗試，JobManager和TaskManager之間關于Task的部署和Task執行狀態的更新都是通過ExecutionAttemptID來標識實體的，在故障或者資料需要重算的情況下，ExecutionVertex可能會有多個ExecutionAttemptID.一個Execution通過ExecutionAttemptID標識，

2.3.2 ExecutionGrap生成程序

初始話作業調度器的時候，根據JobGraph生活ExecutionGraph，在SchedulerBase的構造方法中觸發構建，最終呼叫SchedulerBase#createExecutionGraph 觸發實際的構建動作，使用ExecutionGraphBuiler構建ExecutionGraph，

核心代碼attachJobGraph：

構建ExecutionEdge 的連接策略：

• 點對點連接（DistributionPattern.POINTWISE）

該策略用來連接當前ExecutionVertex與上游的IntermediataeResultParition，

連接分三種情況

1）一對一連接：并發的Task數量與磁區數相等，

2）多對一連接：下游的Task數量小于上游的磁區數，此時分兩種情況：

a：下游Task可以分配同數量的結果磁區IntermediataeResultParition，如上游有4個結果磁區，下游有2個Task，那么每個Task會分配兩個結果磁區進行消費，

b：每個Task消費的上游磁區結果資料不均，如上游有3個結果磁區，下游有兩個Task，那么一個Task分配2個結果磁區消費，另一個分配一個結果磁區消費，

3）一對多連接：下游的Task數量多余上游的磁區數，此時兩種情況：

a：每個結果磁區的下游消費Task資料量相同，如上游有兩個結果磁區，下游有4個Task，每個結果磁區被兩個Task消費，

b：每個結果磁區的下游消費Task數量不相同，如上游有兩個結果磁區，下游有3個Task，那么一個結果磁區分配2個Task消費，另一個結果磁區分配一個Task消費，

• 全連接（DistributionPattern.ALL_TO_ALL）

該策略下游的ExecutionVertex與上游的所有IntermediataeResultParition建立連接，消費其生產的資料，一般全連接的情況意味著資料在Shuffle，

接下來Flink資源管理篇，如果對Flink感興趣或者正在使用的小伙伴，可以加我入群一起探討學習，

參考書籍《Flink 內核原理與實作》

歡迎關注公眾號： 資料基石