Flink-作业提交流程

向右看齐 2022-12-01 15:54 237阅读 0赞

# Flink-作业提交流程 #

# 快速索引 #

1.  StreamGraph  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center]
2.  JobGraph  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 1]
3.  ExecutionGraph
4.  RuntimeGraph

# 1 背景 #

## 1.1 概述 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 2]

Flink DataStream一个作业提交，大致有以下几步

1.  编译阶段-Client生成StreamGraph
    
    1.  将每个算子转换为Transformation，并且记录下每个input Transformation。
    2.  将每个Transformation转换为StreamNode，以及带有并行度和Partitioner信息的、有向的StreamEdge，StreamNode会记录出边outEdges和入边inEdges
    3.  这样就得到了一个有向图StreamGraph
2.  编译阶段-Client通过StreamGraph生成JobGraph  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 3]
    
    1.  将无shuffle的node合并，如上图的Keyed Aggregation和Sink。
    2.  为每个输出生成IntermediateDataset
    3.  将StreamEdge转为JobEdge
3.  运行阶段-将JobGraph提交到集群的JobManager，生成和调度 ExecutionGraph  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 4]  
    JM将客户端提交过来的JobGraph ，将任务按并行度拆分为subtask、IntermediateDataset拆分为IntermediateResultPartition，按Edge上的partitioner信息拆分为`ExecutionEdge`， 最终得到ExecutionGraph
4.  运行阶段-JM将ExecutionGraph概念化得到一张运行图（虚拟图）  
    此时`IntermediateResultPartition`根据下游算子并行度拆分为了ResultSubPartition负责发送数据；
    
    而ExecutionEdge变为`InputChannel`负责接收数据，和上游ResultSubPartition形成对应关系  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 5]

## 1.2 示例代码 ##

import org.apache.flink.api.java.utils.ParameterTool
    import org.apache.flink.streaming.api.scala._
    
    object TextWordCount { 
      def main(args: Array[String]) { 
    
        // Checking input parameters
        val params = ParameterTool.fromArgs(args)
    
        // set up the execution environment
        val env = StreamExecutionEnvironment.getExecutionEnvironment
    
        // make parameters available in the web interface
        env.getConfig.setGlobalJobParameters(params)
    
        // get input data
        val text =
        // read the text file from given input path
          if (params.has("input")) { 
            env.readTextFile(params.get("input"))
          } else { 
            println("Executing WordCount example with default inputs data set.")
            println("Use --input to specify file input.")
            // get default test text data
            env.fromElements(WordCountData.WORDS: _*)
          }
    
        val counts: DataStream[(String, Int)] = text
          // split up the lines in pairs (2-tuples) containing: (word,1)
          .flatMap(p=>p.toLowerCase.split("\\W+"))
          .uid("flatMap-1")
          .filter(_.nonEmpty)
          .uid("filter-1")
    // .map((_, 1))
          .map((p)=>(p,1))
          .uid("map-1")
          // group by the tuple field "0"
          .keyBy(0)
          //sum up tuple field "1"
          .sum(1)
    
        // emit result
        if (params.has("output")) { 
          counts.writeAsText(params.get("output"))
        } else { 
          println("Printing result to stdout. Use --output to specify output path.")
          counts.print()
        }
    
        // execute program
        env.execute("Streaming WordCount")
      }
    }

# 2 编译阶段-StreamGraph的诞生 #

## 2.1 流程 ##

一个DataStream Job提交流程如下：

1.  添加Source  
    `StreamExecutionEnvironment#addSource`方法，会创建表示DataStream起点的`DataStreamSource`，随后构造父类`SingleOutputStreamOperator`和租父类`DataStream`实例。
    
    注意，这里创建几个重要组件:
    
     *  Function，就是各个函数，如`FromElementsFunction`  
        创建`fromElements`对应的`FromElementsFunction` \--implements-> \{SourceFunction --extends-> Function, CheckpointedFunction(可使用和创建状态快照)\}，创建后放在Operator `StreamSource`中。
        
        SourceFunction中定义了一个重要的方法`run(SourceContext<T> ctx)`，用来启动Source，可使用`SourceContext`来发送元素到下游算子。
        
        常用 Source Connector如`FlinkKafkaConsumerBase`也是 --extends-> SourceFunction，他的`run(SourceContext<T> ctx)`方法内创建了一个`KafkaFetcher`，用会调用其`runFetchLoop`方法，开启两个线程，一个`KafkaConsumerThread`用于不断从Kafka消费数据交给`Handover`，另一个线程不断从`Handover`中取出数据，并按分区进行处理（1. `sourceContext.collectWithTimestamp(record, timestamp)`，即发送记录时带上从`kafkaEventTimestamp` 2.尝试发送水位 3.更新分区Offset）
     *  Operator，用来做一些`Function`相关操作  
        创建`StreamSource` \--extends-> AbstractUdfStreamOperator --extends-> AbstractStreamOperator --implements-> \{StreamOperator(保证每个算子实例是单线程执行，还继承了通知Checkpoint结果的`CheckpointListener`) , CheckpointedStreamOperator\}，创建后放在要创建的DataStream `DataStreamSource`中。
        
        这里还定义了`chainingStrategy = ChainingStrategy.HEAD`，表明本算子只能在OperatorChain的头结点。
        
        run方法会调用我们定义的`userFunction#run(SourceContext<T> ctx)`方法
     *  DataStream  
        创建`DataStreamSource<T>` \--extends-> SingleOutputStreamOperator --extends-> DataStream
        
        他表示一个DataStream的起点。泛型`T`表示Source流中流动的元素类型。
     *  Transformation，表示创建该DataStream的算子  
        创建一个表示Source的`LegacySourceTransformation`，存在首个DataStream实例`DataStreamSource`中。
    
    就算是调用`StreamExecutionEnvironment#fromElements`或`StreamExecutionEnvironment#fromCollection`，其实也是调用`addSource`，只不过此时并行度为1，且不能为他设置更高的并行度，因为他们不是可并行算子，否则会报错`IllegalArgumentException: The parallelism of non parallel operator must be 1.`.
2.  定义各个转换算子  
    如`.flatMap(p=>p.toLowerCase.split("\\W+"))`：
    
     *  Function  
        使用用户定义的flatMap函数内容(`flatMap(p=>p.toLowerCase.split("\\W+"))`) 来创建 `FlatMapFunction` \--extends-> Function
     *  Operator  
        `StreamFlatMap`–extends-> AbstractUdfStreamOperator ->…, --implements-> OneInputStreamOperator（定义了重要方法`processElement(StreamRecord<IN> element)`，为每个元素调用该方法，执行我们定义的flatMap Function，结果放入collector）
     *  Transformation  
        先创建`OneInputTransformation`（表示拥有一个输入），存在在`SingleOutputStreamOperator`这个DataStream中。
     *  SingleOutputStreamOperator --extends-> DataStream  
        将DataStream类内该算子及实现内容以及元素类型信息`TypeInformation`（这个例子中是String）作为参数，调用`DataStream#transform`方法进行转换并返回构建的DataStream子类`SingleOutputStreamOperator`。
    
    flatMap函数的上述构建过程如下：
    
        public <R> SingleOutputStreamOperator<R> flatMap(FlatMapFunction<T, R> flatMapper, TypeInformation<R> outputType) { 
        	return transform("Flat Map", outputType, new StreamFlatMap<>(clean(flatMapper)));
        }
    
    随后会调用`DataStraem#doTransform`方法：
    
        OneInputTransformation<T, R> resultTransform = new OneInputTransformation<>(
        		this.transformation,
        		operatorName,
        		operatorFactory,
        		outTypeInfo,
        		environment.getParallelism());
        
        @SuppressWarnings({ "unchecked", "rawtypes"})
        SingleOutputStreamOperator<R> returnStream = new SingleOutputStreamOperator(environment, resultTransform);
        
        getExecutionEnvironment().addOperator(resultTransform);
        
        return returnStream;
    
    1.  先使用本算子信息（名字、工厂类、输出类型、并行度）和上一个算子（如`LegacySourceTransformation`。上个算子的Transformation对象放在上一个DataStream内部，在制作下一个算子的Transformation和DataStream的时候用到）作为`input` Transformation创建一个表示本算子的`Transformation`的子孙类`OneInputTransformation`。
        
        Transformation有一个静态方法`getNewNodeId`用来为每个转换分配唯一ID:
        
            // This is used to assign a unique ID to every Transformation
            protected static Integer idCounter = 0;
            
            public static int getNewNodeId() { 
            	idCounter++;
            	return idCounter;
            }
    2.  然后据此创建`SingleOutputStreamOperator` DataStream（这里面有很多算子相关方法）.
    3.  随后，调用`StreamExecutionEnvironment#addOperator`将本`OneInputTransformation`放入StreamExecutionEnvironment.TransformationList: `transformations`
    4.  如果是存在repartition的算子，如`keyBy`，则会创建`PartitionTransformation`，构建时会传入一个`StreamPartitioner`和`shuffleMode`。
        
        在创建PartitionTransformation后，会据此创建`DataStream`的子类`KeyedStream`，表示这个流按key分配
        
            public KeyedStream(DataStream<T> dataStream, KeySelector<T, KEY> keySelector, TypeInformation<KEY> keyType) { 
            	this(
            		dataStream,
            		new PartitionTransformation<>(
            			dataStream.getTransformation(),
            			new KeyGroupStreamPartitioner<>(keySelector, StreamGraphGenerator.DEFAULT_LOWER_BOUND_MAX_PARALLELISM)),
            		keySelector,
            		keyType);
            }
        
        需要注意的是，该PartitionTransformation不会放入transformations list里面！但会作为下一个聚合函数算子的input被调用`transform`方法。  
        ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 6]
3.  当在定义了一个operator后调用`.uid`方法时，会将该算子对应的DataStream内部的`Transformation.uid`属性重设为用户定义的名字（默认为null）。
4.  后续每个算子添加，都利用前一个算子来创建`OneInputTransformation`和`SingleOutputStreamOperator`，并添加到`StreamExecutionEnvironment.transformations`
5.  对于类似`.sum`这样的聚合计算算子  
    这类算子只能在`KeyedStream`上执行。
    
     *  Function  
        先创建一个`SumAggregator`（–extends-> AggregationFunction --extends-> ReduceFunction --extends-> Function）
     *  Operator  
        使用`SumAggregator`和`TypeSerializer`来创建`StreamGroupedReduce`。他也继承了`AbstractStreamOperator`，也有实现自`OneInputStreamOperator`的`processElement(StreamRecord<IN> element)`方法来调用Function：
        
            @Override
            public void processElement(StreamRecord<IN> element) throws Exception { 
            	IN value = element.getValue();
            	IN currentValue = values.value();
            
            	if (currentValue != null) { 
            		// SumAggregator
            		IN reduced = userFunction.reduce(currentValue, value);
            		// 更新ValueState
            		values.update(reduced);
            		// 发送数据到下游
            		output.collect(element.replace(reduced));
            	} else { 
            		values.update(value);
            		output.collect(element.replace(value));
            	}
            }
     *  Transformation  
        OneInputTransformation，保存在SingleOutputStreamOperator
     *  SingleOutputStreamOperator
6.  addSink  
    比如`DataStream#print`其实就是`addSink(printFunction).name("Print to Std. Out")`.
    
     *  Function  
        此时会先创建`PrintSinkFunction`
     *  Operator  
        然后创建执行`SinkFunction`的`StreamSink`。
        
        该类的上层接口有`OneInputStreamOperator`\->`StreamOperator`，定义了`processElement`（会调用`SinkFunction.invoke(element.getValue(), sinkContext)`）、`processWatermark`等重要方法；
        
        还有一个抽象父类`AbstractUdfStreamOperator`，定义了`initializeState`、`snapshotState`、`notifyCheckpointComplete`、`notifyCheckpointAborted`等状态和Checkpoint相关方法。
     *  Transformation  
        创建Sink对应的转换`SinkTransformation`
     *  DataStreamSink  
        注意这里不再创建`DataStream`，而是根据前面的信息创建`DataStreamSink`，他功能类似DataStream+SingleOutputStreamOperator  
        ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 7]
7.  `StreamExecutionEnvironment#execute`，Job开始真正提交运行
    
        execute(getStreamGraph(jobName));
8.  要先创建`StreamGraph`
    
        public StreamGraph getStreamGraph(String jobName, boolean clearTransformations) { 
        	StreamGraph streamGraph = getStreamGraphGenerator().setJobName(jobName).generate();
        	if (clearTransformations) { 
        		this.transformations.clear();
        	}
        	return streamGraph;
        }
9.  创建`StreamGraphGenerator`  
    这里传入的`transformations`就是我们之前已经加入的表示各个算子的Transformation。
    
        return new StreamGraphGenerator(transformations, config, checkpointCfg)
        		.setStateBackend(defaultStateBackend)
        		.setChaining(isChainingEnabled)
        		.setUserArtifacts(cacheFile)
        		.setTimeCharacteristic(timeCharacteristic)
        		.setDefaultBufferTimeout(bufferTimeout);
10. StreamGraphGenerator\#generate
    
        public StreamGraph generate() { 
        	// 构建一个权限的StreamGraph，这时还没有算子信息
        	streamGraph = new StreamGraph(executionConfig, checkpointConfig, savepointRestoreSettings);
        	// 设置各属性
        	streamGraph.setStateBackend(stateBackend);
        	streamGraph.setChaining(chaining);
        	streamGraph.setScheduleMode(scheduleMode);
        	streamGraph.setUserArtifacts(userArtifacts);
        	streamGraph.setTimeCharacteristic(timeCharacteristic);
        	streamGraph.setJobName(jobName);
        	streamGraph.setGlobalDataExchangeMode(globalDataExchangeMode);
        
        	alreadyTransformed = new HashMap<>();
        	
        	// 将所有表示算子的Transformation进行转换，
        	// 创建StreamNode和StreamEdge，得到StreamGraph
        	// 详见2.2transform(transformation)方法详解
        	for (Transformation<?> transformation: transformations) { 
        		transform(transformation);
        	}
        
        	final StreamGraph builtStreamGraph = streamGraph;
        
        	alreadyTransformed.clear();
        	alreadyTransformed = null;
        	streamGraph = null;
        
        	return builtStreamGraph;
        }
11. 制作好的StreamGraph如下  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center]
12. StreamExecutionEnvironment\#execute(StreamGraph)

## 2.2 StreamGraphGenerator\#transform(transformation) ##

该方法将所有表示算子的Transformation进行转换，创建StreamNode和StreamEdge，得到StreamGraph

1.  根据job设置重设`MaxParallelism`
2.  根据transformation类型，调用对应转换方法  
    比如`transformOneInputTransform`方法，又会递归调用本转换的input进行转换
3.  比如第一个转换是flatMap，则input为`LegacySourceTransformation`，那就继续调用对应的转换方法`transformLegacySource`来转换source
4.  创建表示该实例的StreamNode  
    该`transformXXX`方法会调用`StremaGraph#addOperator`，创建表示该实例的StreamNode。注意，这里就会创建`invokableClass`，如果是Source则创建`class org.apache.flink.streaming.runtime.tasks.SourceStreamTask`；其他算子创建`class org.apache.flink.streaming.runtime.tasks.OneInputStreamTask`
    
    并以id为key，StreamNode为value放入streamNodes Map。
    
    然后根据输入输出类型设置序列化器，还要将sourceId记录到sources Set。最后为该StreamNode设置并行度
5.  创建有向的StreamEdge  
    Source添加完后，继续调用addOperator添加本算子，流程部分和source相同  
    不同的是，因为普通算子有input(如source，而Source没有input上游算子了)，会调用`addEdge`方法来创建上游节点和本节点的`StreamGraph#Edge`。这里就会用到Partitioner
    
     *  在没有指定partitioner时，如果上下游并行度一致，则会使用ForwardPartitioner（一一对应)
     *  否则使用RebalancePartitioner(轮询)
    
    随后使用这些信息创建StreamEdge，这是一个有向边！
    
    StreamNode会记录出边outEdges和入边inEdges
6.  Repartition算子对应的PartitionTransformation的转换处理  
    前面提到过，如果是存在repartition的算子，如`keyBy`，则会为该算子创建`PartitionTransformation`，构建时会传入一个`StreamPartitioner`和`shuffleMode`。
    
    在转换时，PartitionTransformation会按并行度，为每个上游input创建一个虚拟node并保存
    
        virtualPartitionNodes.put(virtualId, new Tuple3<>(originalId, partitioner, shuffleMode));
7.  最后的`SinkTransformation`执行`streamGraph#addSink`保存到streamGraph，
8.  到这里，对所有Transformation的转换阶段结束

# 3 编译阶段-JobGraph的诞生 #

## 3.1 流程 ##

1.  StreamExecutionEnvironment\#execute(StreamGraph)
    
        public JobClient executeAsync(StreamGraph streamGraph) throws Exception { 
        
        	final PipelineExecutorFactory executorFactory =
        		executorServiceLoader.getExecutorFactory(configuration);
        	// 生成JobGraph，并提交执行Job
        	// 利用返回的JobClient来与Job交互，如cancel、savepoint
        	CompletableFuture<JobClient> jobClientFuture = executorFactory
        		.getExecutor(configuration)
        		.execute(streamGraph, configuration);
        
        	try { 
        		JobClient jobClient = jobClientFuture.get();
        		jobListeners.forEach(jobListener -> jobListener.onJobSubmitted(jobClient, null));
        		return jobClient;
        	} catch (ExecutionException executionException) { 
        		final Throwable strippedException = ExceptionUtils.stripExecutionException(executionException);
        		jobListeners.forEach(jobListener -> jobListener.onJobSubmitted(null, strippedException));
        
        		throw new FlinkException(
        			String.format("Failed to execute job '%s'.", streamGraph.getJobName()),
        			strippedException);
        	}
        }
2.  LocalExecutor\#execute  
    核心代码：

final JobGraph jobGraph = getJobGraph(pipeline, effectiveConfig);
    
    return PerJobMiniClusterFactory.createWithFactory(effectiveConfig, miniClusterFactory).submitJob(jobGraph);

1.  远程模式执行会调用`ExecutorUtils#getJobGraph`

// 这里的Pipeline就是StreamGraph
    public static JobGraph getJobGraph(@Nonnull final Pipeline pipeline, @Nonnull final Configuration configuration) { 
    	checkNotNull(pipeline);
    	checkNotNull(configuration);
    	// 用来访问执行相关配置
    	final ExecutionConfigAccessor executionConfigAccessor = ExecutionConfigAccessor.fromConfiguration(configuration);
    	// 生成JobGraph
    	final JobGraph jobGraph = FlinkPipelineTranslationUtil
    			.getJobGraph(pipeline, configuration, executionConfigAccessor.getParallelism());
    	
    	// 配置执行的jar、classpath、savepoint设置等
    	jobGraph.addJars(executionConfigAccessor.getJars());
    	jobGraph.setClasspaths(executionConfigAccessor.getClasspaths());
    	jobGraph.setSavepointRestoreSettings(executionConfigAccessor.getSavepointRestoreSettings());
    	// 到这里，整个JobGraph构建完成
    	return jobGraph;
    }

1.  FlinkPipelineTranslationUtil\#getJobGraph

public static JobGraph getJobGraph(
    		Pipeline pipeline,
    		Configuration optimizerConfiguration,
    		int defaultParallelism) { 
    	// StreamGraphTranslator
    	FlinkPipelineTranslator pipelineTranslator = getPipelineTranslator(pipeline);
    
    	return pipelineTranslator.translateToJobGraph(pipeline,
    			optimizerConfiguration,
    			defaultParallelism);
    }

1.  StreamingJobGraphGenerator\#createJobGraph  
    创建StreamingJobGraphGenerator，然后用他来createJobGraph

public static JobGraph createJobGraph(StreamGraph streamGraph, @Nullable JobID jobID) { 
    	return new StreamingJobGraphGenerator(streamGraph, jobID).createJobGraph();
    }

1.  createJobGraph

private JobGraph createJobGraph() { 
    	// 验证checkpoint相关配置
    	preValidate();
    
    	// 调度模式，EAGER或是LAZY_FROM_SOURCES
    	jobGraph.setScheduleMode(streamGraph.getScheduleMode());
    
    	// Generate deterministic hashes for the nodes in order to identify them across
    	// submission iff they didn't change.
    	// 遍历StreamGraph，为每个StreamNode生成hash，该值会被用作JobVertexID
    	Map<Integer, byte[]> hashes = defaultStreamGraphHasher.traverseStreamGraphAndGenerateHashes(streamGraph);
    
    	// Generate legacy version hashes for backwards compatibility
    	List<Map<Integer, byte[]>> legacyHashes = new ArrayList<>(legacyStreamGraphHashers.size());
    	for (StreamGraphHasher hasher : legacyStreamGraphHashers) { 
    		legacyHashes.add(hasher.traverseStreamGraphAndGenerateHashes(streamGraph));
    	}
    	// 从Source节点开始，将配置应用到链中
    	// 递归创建所有JobVertex实例
    	// 将能合并的StreamNode合并为1个JobVertex，详见3.2
    	setChaining(hashes, legacyHashes);
        // 详见3.4
    	setPhysicalEdges();
    	// 详见3.5
    	setSlotSharingAndCoLocation();
    	// 为每个算子设置`ManagedMemory`内存比例
    	setManagedMemoryFraction(
    		Collections.unmodifiableMap(jobVertices),
    		Collections.unmodifiableMap(vertexConfigs),
    		Collections.unmodifiableMap(chainedConfigs),
    		id -> streamGraph.getStreamNode(id).getMinResources(),
    		id -> streamGraph.getStreamNode(id).getManagedMemoryWeight());
    	// 详见3.6
    	configureCheckpointing();
    
    	jobGraph.setSavepointRestoreSettings(streamGraph.getSavepointRestoreSettings());
    
    	JobGraphUtils.addUserArtifactEntries(streamGraph.getUserArtifacts(), jobGraph);
    
    	// set the ExecutionConfig last when it has been finalized
    	try { 
    	    // 详见3.7
    		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;
    }

1.  得到的JobGraph  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 1]  
    可以看到，之前的StreamGraph中的6个StreamNode在JobGraph中经过合并转换变为了3个jobVertex。
    
    其中：
    
     *  Source因为其输出边的Partitioner为RebalancePartitioner所以是单独的，原因是我们用例中采用了`env#fromElements`来构建source，这是一个非并行算子，并行度强行置为1。但如果使用如`FlinkKafkaConsumer`这类的Source，则可以和后续算子并行度保持一致，使用`ForwardPartitioner`
     *  而FlatMap、Filter、Map合并为一  
        Aggregation的`ChainingStrategy`为`HEAD`所以不能合并到Map；Map输出边的的Partitioner是`HASH`，也不满足
     *  Aggregation和Sink合并为一
    
    当我们改用`FlinkKafkaConsumer`作为Source时得到的JobGraph如下图：  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 8]  
    可以看到因为此时Source和FlatMap算子并行度一致，所以组成一个JobVertex。

## 3.2 createChain ##

StreamingJobGraphGenerator\#createChain  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 9]  
以上为入度为1，出度为2的一个OperatorChain示例。

List<StreamEdge> createChain(Integer currentNodeId, int chainIndex, OperatorChainInfo chainInfo)

这个方法是构成JobGraph的核心步骤，返回的内容是`List<StreamEdge> transitiveOutEdges`，部分代码和说明说下：

1.  判断当前StreamNode的输出的StreamEdge是否能构成chain，可以（如FlatMap和Filter、Map；Aggregation和Sink）就将该边放入`chainableOutputs` List，否则（如Source和FlatMap）放入`nonChainableOutputs` List
2.  判断是否构成Chain的条件很苛刻，需要以下提交件全部都为true:
    
     *  下游StreamNode的入边数为1，即只有一个input，而不是多个输入
     *  上游StreamNode和下游StreamNode同在一个`SlotSharingGroup`
     *  上游的`ChainingStrategy`不能是`NEVER`  
        即只能是`ALWAYS`（可参与上下游连接，如FlatMap、Filter、Map）或`HEAD`（只能作为上游与下游连接，但不能与下游连接，如Source）
     *  下游必须是`ALWAYS`
     *  该StreamEdge的Partitioner必须是`ForwardPartitioner`，即发送和接收分区一一对应
     *  该StreamEdge的ShuffleMode不能为BATCH，该模式下生产者先生产完整数据，然后消费者才开始消费数据；还有个模式是PIPELINED，此时生产者消费者同时完成生产消费；UNDEFINED的含义是由Flink推断到底采用以上哪种模式。
     *  上下游并行度相同
     *  StreamGraph允许Chain机制。默认true，可通过`pipeline.operator-chaining`设置。
3.  对于`chainableOutputs`里的所有StreamEdge指向的StreamNode，递归调用`createChain(chainable.getTargetId(), chainIndex + 1, chainInfo))`方法，将结果（递归调用产生的transitiveOutEdges）放入`transitiveOutEdges` list
4.  对于`nonChainableOutputs`（如Source->FlatMap、Map->Aggregation partitioner为HASH）里的所有StreamEdge放入`transitiveOutEdges` list，并递归调用`createChain(nonChainable.getTargetId(), 0, chainInfo.newChain(nonChainable.getTargetId()));`方法
5.  创建ChainName和OperatorName的映射的Map `chainedNames`
6.  将当前算子信息放入该`OperatorChainInfo`的`chainedOperatorHashes`，以备同Chain的头结点合并该Chain上的StreamNode并创建jobVertex时使用。
7.  **为同一个OperatorChainInfo的成Chain的若干StreamNode节点创建为一个`jobVertex`，放入`jobGraph.jobVertices`**  
    这一步很关键，就已经将可以连成Chain的算子的StreamNode合并为jobVertex，内部包含所有算子信息
8.  创建`StreamConfig`，为当前节点设置各个属性，存入`vertexConfigs` Map
9.  如果当前节点为该OperatorChainInfo的头结点(`currentNodeId.equals(startNodeId)`)，则会将当前收集到的`nonChainableOutputs`内的StreamEdge调用`connect(startNodeId, edge)`方法，详见3.3

## 3.3 connect ##

StreamingJobGraphGenerator\#connect

本方法的作用是连接不同的OperatorChain，创建上游输出`IntermediateDataSet`和连接上下游输入的边`JobEdge`

1.  到这个方法的StreamEdge说明是不能组成Chain的节点之间的边，会放入`physicalEdgesInOrder`
2.  确定`ResultPartitionType`，默认为`ResultPartitionType.PIPELINED_BOUNDED`
3.  为上游的JobVertex `headVertex`（即上游节点）创建`IntermediateDataSet`（中间数据集），并放入一个list `JobVertex.results`保存  
    IntermediateDataSet表示由一个算子（可能是Source或任何中间算子）生产的数据集，包含了`ResultPartitionType`、Producer（就是该JobVertex）等信息
    
    中间数据集中的数据可能由其他算子读取、物化或抛弃。
    
    ResultPartitionType是`ResultPartition`的partition模式，根据JobEdge的ShuffleMode有不同选项：
    
     *  ShuffleMode.PIPELINED  
        此时生产者消费者同时完成生产消费。
        
        此时ResultPartition使用`ResultPartitionType.PIPELINED_BOUNDED`。表示拥有本地有界BufferPool的Pipelined partition。
        
         *  对于流式作业来说，这种缓存上界是有必要的，可避免过多数据被缓存，导致CheckpointBarrier延迟。没有限制整体的NetworkBufferPool大小，只限制本地BufferPool可以通过选择适当的大网络缓冲池大小，允许在分区总数方面保持灵活性。
         *  对于批作业来说，最好是选择无限制的`PIPELINED`模式，因为不需要CheckpointBarrier
     *  ShuffleMode.BATCH  
        该模式下生产者先生产完整数据，然后消费者才开始消费数据。
        
        此时ResultPartition使用`ResultPartitionType.BLOCKING`。表示阻塞式的数据交换，即数据流先被完全生产出来，然后再被消费者消费。仅适用于有界流，可以被用在有界流运行时和恢复时。这类数据可被并行地消费多次。这类分区不会在被消费后被自动释放，只能有Scheduler决定该分区不再被需要后被释放。
     *  ShuffleMode.UNDEFINED  
        由Flink推断到底采用以上哪种模式。
        
        此时ResultPartition使用`streamGraph.getGlobalDataExchangeMode()`进行推断：
        
         *  默认为`ALL_EDGES_PIPELINED`  
            `ResultPartitionType.PIPELINED_BOUNDED`
         *  ALL\_EDGES\_BLOCKING  
            `ResultPartitionType.BLOCKING`
         *  FORWARD\_EDGES\_PIPELINED  
            如果`edge.getPartitioner()`为`ForwardPartitioner`，则使用`ResultPartitionType.PIPELINED_BOUNDED`；否则`ResultPartitionType.BLOCKING`
         *  POINTWISE\_EDGES\_PIPELINED  
            如果`edge.getPartitioner()`为`ForwardPartitioner`或`RescalePartitioner`，则使用`ResultPartitionType.PIPELINED_BOUNDED`；否则`ResultPartitionType.BLOCKING`
4.  在上下游JobVertex之间创建`JobEdge`，连接上游的IntermediateDataSet和下游的JobVertex
    
     *  如果当前StreamEdge的partitioner为ForwardPartitioner或RescalePartitioner，则`DistributionPattern`为`POINTWISE`，即每个ProducerSubtask只需要连接到一个或多个ConsumerSubtask。
     *  其他情况，则`DistributionPattern`为`ALL_TO_ALL`，即每个ProducerSubtask都连接到ConsumerSubtask。
5.  JobEdge保存在上游`JobVertex.resuluts[i].consumers`以及下游`JobVertex.inputs`中

## 3.4 setPhysicalEdges ##

StreamingJobGraphGenerator\#setPhysicalEdges

遍历3.3 connect中放入`physicalEdgesInOrder`的StreamEdge，将该边的下游的JobVertex对应的输入InPhysicalEdge设为该物理StreamEdge

## 3.5 setSlotSharingAndCoLocation ##

StreamingJobGraphGenerator\#setSlotSharingAndCoLocation

### 3.5.1 setSlotSharing ###

为所有JobVertex创建`SlotSharingGroup`，SlotSharingGroup**软定义**了不同的task(`JobVertex`)是否可在一个Slot中运行。

1.  先构建Region  
    因为我们使用默认配置的DataStreamApi，所以这里只会有一个包含三个JobVertex的Region
2.  设置SlotSharingGroup  
    三个JobVertex都设置了相同的SlotSharingGroup，因为使用默认SharingGroup `default`

### 3.5.2 setCoLocation ###

`CoLocationGroup`是JobVertex组，**硬性**定义了某个JobVertex的第i个子任务必须和其他所有JobVertex的第i个子任务运行在相同TaskManager上。

这个例子中，没有设置`coLocationGroup`，所以没有做什么操作

## 3.6 StreamingJobGraphGenerator\#configureCheckpointing ##

Checkpoint相关配置，构建几个核心对象

*  triggerVertices  
    触发Checkpoint时的执行起点JobVertex
 *  ackVertices  
    触发Checkpoint时的执行时的所有需要ack的JobVertex
 *  commitVertices  
    触发Checkpoint时的执行时的所有需要commit的JobVertex
 *  checkpointCoordinatorConfiguration  
    构建CheckpointCoordinator所需配置。  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 10]

## 3.7 jobGraph\#setExecutionConfig ##

为JobGraph设置运行配置，直接从StreamGraph取就行了。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 11]

# 4 提交阶段-JobGraph的提交 #

## 4.1 LocalExecutor ##

### 4.1.1 execute ###

@Override
    public CompletableFuture<JobClient> execute(Pipeline pipeline, Configuration configuration) throws Exception { 
    	checkNotNull(pipeline);
    	checkNotNull(configuration);
    
    	Configuration effectiveConfig = new Configuration();
    	effectiveConfig.addAll(this.configuration);
    	effectiveConfig.addAll(configuration);
    
    	// we only support attached execution with the local executor.
    	checkState(configuration.getBoolean(DeploymentOptions.ATTACHED));
    	// pipeline就是StreamGraph
    	// 得到JobGraph
    	final JobGraph jobGraph = getJobGraph(pipeline, effectiveConfig);
    	// 先构建PerJobMiniClusterFactory
    	// 然后提交jobGraph
    	return PerJobMiniClusterFactory.createWithFactory(effectiveConfig, miniClusterFactory).submitJob(jobGraph);
    }

### 4.1.2 submitJob ###

PerJobMiniClusterFactory\#submitJob

public CompletableFuture<JobClient> submitJob(JobGraph jobGraph) throws Exception { 
    	// 
    	MiniClusterConfiguration miniClusterConfig = getMiniClusterConfig(jobGraph.getMaximumParallelism());
    	MiniCluster miniCluster = miniClusterFactory.apply(miniClusterConfig);
    	// 启动一个本地的miniCluster
    	miniCluster.start();
    	// 往这个本地集群提交jobGraph作业
    	return miniCluster
    		.submitJob(jobGraph)
    		.thenApply(result -> new PerJobMiniClusterJobClient(result.getJobID(), miniCluster))
    		.whenComplete((ignored, throwable) -> { 
    			if (throwable != null) { 
    				// We failed to create the JobClient and must shutdown to ensure cleanup.
    				shutDownCluster(miniCluster);
    			}
    		})
    		.thenApply(Function.identity());
    }

### 4.1.3 Ministart\#start ###

#### 4.1.3.1 start ####

1.  创建AkkaRpcService
2.  创建JM和TM用来处理阻塞IO操作的`ioExecutor`线程池  
    核心线程默认为`Runtime.getRuntime().availableProcessors()`\*4，也可`Runtime.getRuntime().availableProcessors()`设置
3.  创建HA服务`HighAvailabilityServices`  
    ![在这里插入图片描述][20200902164804745.png_pic_center]
4.  创建用于接收二进制大对象请求的`BlobServer`线程，是一个基于Socket的服务  
    监听相关请求，分配线程处理，创建目录存储BLOB文件或缓存他们
5.  创建用于心跳接收或请求的`HeartbeatServices`
6.  运行用于Blob缓存的`BlobCacheService`（集群模式下在TM的入口类`TaskManagerRunner`中创建）
7.  启动TaskManager  
    本地环境默认只有1个TaskManager，可通过`local.number-taskmanager`修改
    
        // 详见4.1.3.2
        final TaskExecutor taskExecutor = TaskManagerRunner.startTaskManager(
        	configuration,
        	new ResourceID(UUID.randomUUID().toString()),
        	taskManagerRpcServiceFactory.createRpcService(),
        	haServices,
        	heartbeatServices,
        	metricRegistry,
        	blobCacheService,
        	useLocalCommunication(),
        	ExternalResourceInfoProvider.NO_EXTERNAL_RESOURCES,
        	taskManagerTerminatingFatalErrorHandlerFactory.create(taskManagers.size()));
        // 启动taskExecutor的RpcServer，开始接收处理远程过程调用
        taskExecutor.start();
        // 将该taskExecutor加入到taskManagers集合
        taskManagers.add(taskExecutor);
8.  创建Dispatcher、ResourceManager、WebMonitorEndpoint  
    详见4.1.3.3
9.  分别创建RpcGatewayRetriever Listener，传递给HighAvailabilityServices中获取的三个Retriever，并启动
    
        resourceManagerLeaderRetriever.start(resourceManagerGatewayRetriever);
        dispatcherLeaderRetriever.start(dispatcherGatewayRetriever);
        clusterRestEndpointLeaderRetrievalService.start(webMonitorLeaderRetriever);
10. 开始提交job流程，详见4.1.4

#### 4.1.3.2 TaskManagerRunner\#startTaskManager ####

注意：

*  此时尚未启动。
 *  因为是Local模式，所以才会调用该方法来启动TaskManager

1.  从用户配置获取TaskExecutor可用资源`TaskExecutorResourceSpec`  
    ![在这里插入图片描述][20200916112040650.png_pic_center]
2.  创建`taskManagerServicesConfiguration`  
    ![在这里插入图片描述][20200916112421794.png_pic_center]
3.  创建TM用来处理阻塞IO操作的`ioExecutor`线程池  
    核心线程默认为`Runtime.getRuntime().availableProcessors()`\*4，也可`Runtime.getRuntime().availableProcessors()`设置
4.  创建`TaskManagerServices`  
    他是`TaskExecutor`服务（如`MemoryManager`、`IOManager`、`ShuffleEnvironment`、TaskSlotTable(存放本TM管理的多个TaskSlot实例以及Slot的CPU和内存资源量，这里就已经初始化现有Slot资源)、JobTable（管理运行在本TM上的Job的生命周期）等）的容器，这些服务是每个TaskExecutor实例独享的。
5.  创建`TaskManagerConfiguration`  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 12]
6.  创建可运行多个Task的TaskExecutor  
    他继承自`RpcEndpoint`，通过底层AkkaRpcService来实现远程过程调用  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 13]
7.  构建TaskExecutor父类RpcEndpoint
    
        protected RpcEndpoint(final RpcService rpcService, final String endpointId) { 
        	// AkkaRpcService
        	this.rpcService = checkNotNull(rpcService, "rpcService");
        	this.endpointId = checkNotNull(endpointId, "endpointId");
        	// 开启基于Akka通信的RpcServer
        	// 会创建一个动态代理AkkaInvocationHandler，去使用Akka
        	this.rpcServer = rpcService.startServer(this);
        	// 创建在主线程上下文执行线程任务的执行器
        	this.mainThreadExecutor = new MainThreadExecutor(rpcServer, this::validateRunsInMainThread);
        }
8.  一旦RpcEndpoint启动，就会调用`onStart`方法  
    将`ResourceManagerLeaderListener`注册到`EmbeddedLeaderRetrievalService`，然后启动taskSlotTable、jobLeaderService（用来寻找JM Leader并获得通知，注册了`JobLeaderListenerImpl`，详见6.1.2.3）、FileCache（用来访问那些当Task部署时注册的、由`PermanentBlobService`管理的缓存文件）
9.  TM查找到ResourceManager Leader地址后，就会用监听器回调`TaskExecution#notifyOfNewResourceManagerLeader`，在这里就使用得到的RML Akka连接地址进行带超时的重连接，会打出日志如下：
    
        Connecting to ResourceManager akka.tcp://flink@192.1.1.1:1111/user/rpc/resourcemanager_*(00000000000000000000000000000000).
10. RetryingRegistration\#startRegistration  
    连接成功后，TM继续注册到RM，会调用`ResourceManagerGateway#registerTaskExecutor`，有一个动态代理，经过RPC调用来注册TM到RM，RM侧对应方法为`ResourceManager#registerTaskExecutor`，注册相关日志如下：
    
        Resolved ResourceManager address, beginning registration
        
        Successful registration at resource manager akka.tcp://flink@192.1.1.1:1111/user/rpc/resourcemanager_* under registration id e82aa140ab02db8c31aa6416217f577d.
11. 注册成功后，会调用`ResourceManagerRegistrationListener#onRegistrationSuccess`，这里就会使用RPC将TaskExecutor携带的Slot资源量传递到RM，详见4.1.3.4
12. 到这里，TM已经成功连接并注册到RM，但需要注意的是，TM与JM的连接和注册要等到RM向TM提起Slot资源分配请求阶段，详见 6.1.2.2 步骤5
13. 等待JM申请Slot，详见6.1.2.2

#### 4.1.3.3 setupDispatcherResourceManagerComponents ####

Ministart\#setupDispatcherResourceManagerComponents

方法名字起的真够长的，不过也说明了大概含义，这一步会创建`DispatcherResourceManagerComponent`，他在一个进程内同时运行Dispatcher、ResourceManager、WebMonitorEndpoint

1.  创建WebMonitorEndpoint  
    其父类是RestServerEndpoint，以Netty为基础实现了一个HTTP RestServer，对外提供Flink Web UI服务
2.  创建ResourceManager
    
    1.  先创建管理Slot资源的`SlotManager`实现类`SlotManagerImpl`，职责是
        
         *  管理所有TM向JM注册的Slot的分配
         *  管理Slot申请请求 `SlotRequest`  
            每当一个Slot注册或之前分配的Slot释放，则会尝试满足`PENDING`状态的slot请求
         *  每当可用Slot资源不足时，通知ResourceManager（ResourceActions\#allocateResource(WorkerResourceSpec)）
         *  为了释放资源并避免资源泄漏，空闲的TM（当前未使用其Slot的TM）和挂起的Slot请求超时分别触发了它们的释放和失败。
    2.  JobLeaderIdService
    3.  创建ResourceManager，因为我们是本地模式，所以这里创建的是`StandaloneResourceManager`，on yarn模式下采用的是`YarnResourceManager`  
        ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 14]
        
         *  RpcEndpoint  
            说明这是一个通过AkkaRpcService支持的Rpc远程调用服务
         *  FencedRpcEndpoint  
            说明这是一个受防护的RpcEndpoint，他期望所有收到的Rpc消息都带有令牌。 而该端点也已分配了自己的令牌， 只有在RPC请求携带的令牌等于请求端点的令牌时，才执行该Rpc请求。
         *  ResourceManager  
            负责资源分配、回收、租借。主要的RPC方法如下：
            
             *  registerJobManager(JobMasterId, ResourceID, String, JobID, Time)  
                将一个Job对应的一个`JobMaster`注册到ResourceManager
             *  requestSlot(JobMasterId, SlotRequest, Time)  
                从ResourceManager申请一个slot
         *  StandaloneResourceManager  
            这个RM不会申请新的资源
3.  创建存储已完成的Job的运行时ExecutionGraph归档的HistoryServerArchivist
4.  创建 `DefaultDispatcherRunner`
5.  启动ResourceManager的Akka RpcServer
6.  返回封装了以上组件的`DispatcherResourceManagerComponent`

#### 4.1.3.4 ResourceManager\#sendSlotReport ####

Flink RM收到TM的注册消息后，会有一个动态代理`AkkaInvocationHandler`（详见4.1.3.2 步骤10），触发本方法。

1.  slotManager\#registerTaskManager  
    将一个TM注册到SlotManager，此后该TM上的Slot资源可由JM进行分配
    
    实际上就是注册到SlotManagerImpl，会将这些新注册的Slot先尝试从`SlotManagerImpl.pendingSlots`中匹配已有的Slot请求；否则就将这个状态为`TaskManagerSlot.State.FREE`的Slot放入`freeSlots`中保存，表示为可分配的空闲Slot资源。
2.  如果TM连接断开，则会移除该TM注册的Slot。
3.  JM连上ResourceManager后会为负责的Job申请Slot资源，详见步骤2.5

### 4.1.4 Ministart\#submitJob(jobGraph) ###

1.  使用BlobClient将 JobGraph相关的UserJar上传
2.  使用`CompletableFuture`来调用`Dispatcher#submitJob(JobGraph jobGraph, @RpcTimeout Time timeout)`
3.  最后返回`CompletableFuture<JobSubmissionResult>`

### 4.1.5 Dispatcher\#submitJob ###

如果是首次提交则生成Dispatcher，他的职责是接受Job提交、持久化Job、产生JobMaster来执行该Job、在JobMaster失败的时候重启Job，随时掌控Flink Session Cluster状态。

1.  构建Dispatcher实现类和父类Dispatcher  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 15]
    
     *  Flink Local模式和Session模式都使用`StandaloneDispatcher`，会在同一个进程中为每次提交的JobGraph生成一个新的JobMaster来管理；
     *  Flink Per-job模式使用`MiniDispatcher`，在初始化Dispatcher的时候只传入单个JobGraph。如果执行模式为`ExecutionMode#DETACHED`，则在Job完成后MiniDispatcher也会立刻结束
2.  启动Dispatcher RpcServer，准备接收JobGraph来执行
3.  通过AkkaRpcService 动态代理`AkkaInvocationHandler`，调用`Dispatcher#submitJob`方法，内部调用`Dispatcher#internalSubmitJob`来提交JobGraph
4.  将该JobGraphId和`createJobManagerRunner`方法返回的CompletableFuture记录到`jobManagerRunnerFutures`
5.  createJobManagerRunner方法创建`JobManagerRunner`
    
    1.  创建ShuffleMaster，管理上游ResultPartition和下游InputGate的Shuffle，目前只有一个实现类`NettyShuffleMaster`
    2.  **创建JobMaster**  
        每个JobMaster只负责单个JobGraph的运行  
        ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 16]  
        会打出日志类似`Initializing job insert into flink_meta.test_sink_uv_mysql_20200820`。主要流程如下：
        
        1.  创建`SlotPool`的实现类`SlotPoolImpl`
            
             *  作用是服务由`ExecutionGraph`向本JM提起的Slot请求，当没有足够Slot时就向ResourceManager申请新的Slot。
             *  一旦申请到Slot，SlotPool也会保存他们，这样即使ResourceManager挂了，依然可以分配这里已空闲的Slot。
             *  当Slot无人使用时，会释放。
             *  所有Slot分配由一个自增的`AllocationID`标记
        2.  创建`Scheduler`的实现类`SchedulerImpl`  
            是`SlotProvider`和`SlotOwner`的实现类。职责是调度Task到Slot上执行。
        3.  创建追踪反压的`BackPressureStatsTrackerImpl`
        4.  (步骤5.2.4)创建调度Flink Job的`SchedulerNG`实现类`DefaultScheduler`，创建父类`SchedulerBase`时会使用JobGraph生成`ExecutionGraph`及其调度拓扑`SchedulingTopology`（详见5.1）
            
            注意：这里的调度器和SchedulerImpl不同，本类看起来是用来调度整个Job ExecutionGraph的  
            还会：
            
            1.  创建 InputsLocationsRetriever  
                创建查找输入的位置点的InputsLocationsRetriever\`
                
                    this.inputsLocationsRetriever = new ExecutionGraphToInputsLocationsRetrieverAdapter(executionGraph);
            2.  创建管理每个Operator算子Id对应的`OperatorCoordinatorHolder`的Map。
        5.  继续创建DefaultScheduler：
            
            1.  会利用刚才创建好的`ExecutionGraph.schedulingTopology`来创建failover策略类`FailoverStrategy`的实现类，比如这里为`RestartPipelinedRegionFailoverStrategy`，即当某个节点失败时，重启一个Region的所有节点。
            2.  创建调度策略`SchedulingStrategy`实现类，如`EagerSchedulingStrategy`，这个实现类表示在所有task在同一时间一起调度
        6.  创建监控心跳的`taskManagerHeartbeatManager`和`resourceManagerHeartbeatManager`
6.  (步骤6)调用`JobManagerRunner#start`启动JobMaster，详见6.1.1.2

### 4.1.6 运行阶段 ###

见6.1

## 4.2 AbstractJobClusterExecutor ##

### 4.2.1 execute ###

per-job on yarn模式会走`AbstractJobClusterExecutor#execute`方法。

@Override
    public CompletableFuture<JobClient> execute(@Nonnull final Pipeline pipeline, @Nonnull final Configuration configuration) throws Exception { 
    	// 创建JobGraph
    	final JobGraph jobGraph = PipelineExecutorUtils.getJobGraph(pipeline, configuration);
    	// 创建描述Yarn集群信息的ClusterDescriptor
    	try (final ClusterDescriptor<ClusterID> clusterDescriptor = clusterClientFactory.createClusterDescriptor(configuration)) { 
    		final ExecutionConfigAccessor configAccessor = ExecutionConfigAccessor.fromConfiguration(configuration);
    		// 要启动的Flink集群的内存和slot配置
    		final ClusterSpecification clusterSpecification = clusterClientFactory.getClusterSpecification(configuration);
    		// 将JobGraph提交到Yarn集群，启动Flink Cluster
    		// 返回一个ClusterClientProvider
    		// 可从他得到持有该集群相关信息如appid等的RestClusterClient
    		final ClusterClientProvider<ClusterID> clusterClientProvider = clusterDescriptor
    				.deployJobCluster(clusterSpecification, jobGraph, configAccessor.getDetachedMode());
    		LOG.info("Job has been submitted with JobID " + jobGraph.getJobID());
    		// 返回一个已经完成的CompletableFuture，内部返回值为JobClient实现类
    		return CompletableFuture.completedFuture(
    				new ClusterClientJobClientAdapter<>(clusterClientProvider, jobGraph.getJobID()));
    	}
    }

### 4.2.2 deployJobCluster ###

per-job on yarn模式会走`YarnClusterDescriptor#deployJobCluster`方法。

@Override
    public ClusterClientProvider<ApplicationId> deployJobCluster(
    	ClusterSpecification clusterSpecification,
    	JobGraph jobGraph,
    	boolean detached) throws ClusterDeploymentException { 
    	try { 
    		return deployInternal(
    			clusterSpecification,
    			"Flink per-job cluster",
    			// 得到在yarn container跑的第一个类YarnJobClusterEntrypoint
    			// ，yarn会运行其main方法
    			getYarnJobClusterEntrypoint(),
    			jobGraph,
    			detached);
    	} catch (Exception e) { 
    		throw new ClusterDeploymentException("Could not deploy Yarn job cluster.", e);
    	}
    }

deployInternal方法部分代码步骤如下：

1.  对配置参数做基本检查，不对就抛出异常  
    isReadyForDeployment(clusterSpecification);
2.  检测目标队列是否存在  
    checkYarnQueues(yarnClient);
3.  为该新Application创建`YarnClientApplication`，得到appId、最大/最小资源容量
    
        @Override
        public YarnClientApplication createApplication()
            throws YarnException, IOException { 
          ApplicationSubmissionContext context = Records.newRecord
              (ApplicationSubmissionContext.class);
          // 向Yarn ResourceManager申请分配一个`ApplicationId` 
          // 返回ApplicationId、最大/最小资源容量
          GetNewApplicationResponse newApp = getNewApplication();
          ApplicationId appId = newApp.getApplicationId();
          context.setApplicationId(appId);
          return new YarnClientApplication(newApp, context);
        }
    
    ApplicationSubmissionContext表示Yarn RM为新应用启动ApplicationMaster所需的所有信息，包括：
    
     *  ApplicationId
     *  用户
     *  应用名称
     *  应用优先级
     *  运行AM的`ContainerLaunchContext`  
        包含ContainerId、分配给该Container的Resource、该Container分配的对象用户、安全token、本地资源（运行container所需，如二进制码、jar、共享对象、其他文件等）、启动进程所需环境变量、**运行container的命令**
4.  从Yarn获取各NM节点可用资源信息及container资源限制信息，判断当前申请的量是否能满足  
    如果不满足则抛出异常，如果满足就返回一个合法的`ClusterSpecification`
    
        return new ClusterSpecification(
        		masterMemoryMB,
        		taskManagerMemoryMB,
        		slotsPerTaskManager);
5.  将组装好的相关信息调用`startAppMaster`来在yarn上启动AppMaster，详见4.2.3
    
        ApplicationReport report = startAppMaster(
        		flinkConfiguration,
        		applicationName,
        		yarnClusterEntrypoint,
        		jobGraph,
        		yarnClient,
        		yarnApplication,
        		validClusterSpecification);
6.  detached模式会打印相关信息，以便让用户得知
7.  将已经运行起来的Application相关信息放入flinkConfiguration中
    
        final ApplicationId clusterId = report.getApplicationId();
        final String host = report.getHost();
        final int port = report.getRpcPort();
        
        flinkConfiguration.setString(JobManagerOptions.ADDRESS, host);
        flinkConfiguration.setInteger(JobManagerOptions.PORT, port);
        
        flinkConfiguration.setString(RestOptions.ADDRESS, host);
        flinkConfiguration.setInteger(RestOptions.PORT, port);
        
        flinkConfiguration.set(YarnConfigOptions.APPLICATION_ID, ConverterUtils.toString(clusterId));
8.  最后返回可获取包含flinkConfiguration的RestClusterClient的`ClusterClientProvider<ApplicationId>`
    
        return () -> { 
        	try { 
        		return new RestClusterClient<>(flinkConfiguration, report.getApplicationId());
        	} catch (Exception e) { 
        		throw new RuntimeException("Error while creating RestClusterClient.", e);
        	}
        };

### 4.2.3 startAppMaster ###

这里再具体看看`YarnClusterDescriptor#startAppMaster`方法主要步骤（这个方法代码超级长！）：

1.  获取`providedLibDirs`目录  
    该路径由`yarn.provided.lib.dirs`指定，此乃HDFS上共享目录，比如flink/lib下的某些jar等，Yarn也会在节点上缓存这些jar，由此可以加速flink job提交和运行速度，不需要每次都去下载了
2.  创建代码上传工具类`YarnApplicationFileUploader`
3.  组装需要被上传到hdfs并会添加到我们app classpath的systemShipFiles
4.  设置zkNamespace和HA  
    如果开启了HA，则会将`ApplicationSubmissionContext.maxAppAttempts`设为`yarn.application-attempts`的值，如没有配设为默认值2；没开启就设为`yarn.application-attempts`的值，如没有配设为默认值1
5.  上传各个jar
    
    1.  组装userJarFiles  
        来源是JobGraph内的UserJars和`pipeline.jars`定义的用户UDF以及依赖的jar
    2.  将`providedLibDirs`内的文件可见性标记为`PUBLIC`范围，并作为应用本地资源
    3.  将`systemShipFiles`和`userJarFiles`内文件的文件上传，将他们的可见性标记为`APPLICATION`范围，并作为应用本地资源
    4.  将`yarn.flink-dist-jar`内的文件上传，并作为应用本地资源
6.  将`JobGraph`对象写入以appId作为前缀的临时文件，随后将该文件也上传，并作为本Application的资源
7.  将systemClassPaths（包括providedLibDirs、systemShipFiles、userClassPaths（是否包含由`userClassPaths`指定））、userClassPaths、yarn.flink-dist-jar、jobGraphFilename
8.  将flink app 配置文件写入临时文件，文件名格式为`appId + "-flink-conf.yaml"`，随后将该文件也上传，并作为本Application的资源
9.  一些kerberos安全相关的操作
10. 根据`jobmanager.memory.process.size`配置指定Flink各个区域的内存资源
11. 调用`YarnClusterDescriptor#setupApplicationMasterContainer` 来创建`ContainerLaunchContext`  
    这里面就会将启动AM时的相关参数组装为启动`YarnJobClusterEntrypoint`的命令
12. 将前面注册的本地资源放入ContainerLaunchContext保存
13. 为ContainerLaunchContext设置运行环境变量，如`_FLINK_CLASSPATH`、`_APP_ID`、`HADOOP_USER_NAME`等
14. 为ApplicationSubmissionContext设置各项参数：
    
        appContext.setApplicationName(customApplicationName);
        appContext.setApplicationType(applicationType != null ? applicationType : "Apache Flink");
        // 就是我们前面提到的ContainerLaunchContext
        appContext.setAMContainerSpec(amContainer);
        // 包含内存和VCore
        appContext.setResource(capability);
        appContext.setPriority(priority);
        // 设定为用户设的queue
        appContext.setQueue(yarnQueue);
        // 配置yarn label
        setApplicationNodeLabel(appContext);
        
        setApplicationTags(appContext);
15. 配置`DeploymentFailureHook`，在部署flink app失败时kill yarn app
16. 提交flink app到yarn，会阻塞到成功提交即被RM接受后才会返回对应的`ApplicationId`  
    底层调用`ApplicationClientProtocol#submitApplication(SubmitApplicationRequest)`
17. 获取`ApplicationReport`应用状态报告  
    通过死循环调用`yarnClient.getApplicationReport(appId)`得到得知yarn app状态为`RUNNING`后，移除DeploymentFailureHook，返回该ApplicationReport，内容包括：  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 17]
18. 接 4.2.2

### 4.2.4 客户端提交Job小结 ###

Per-job on Yarn模式下提交Flink Job，其实就是将相关jar（这其中就包括了我们的JobGraph，也是被写入本地临时文件，作为本地资源的Key为`job.graph`）上传到hdfs并注册为本地资源，将相关配置、启动信息放入`ApplicationSubmissionContext`（内部包含了应用信息及运行AM的`ContainerLaunchContext`），最后作为AMContainer提交到Yarn运行即可。

该模式下，Flink在Yarn上作为AM跑的入口类为`YarnJobClusterEntrypoint`，详见5.2。

## 4.3 AbstractSessionClusterExecutor\#execute ##

### 4.3.1 execute ###

Flink session on yarn模式会走`AbstractSessionClusterExecutor#execute`方法。

@Override
    public CompletableFuture<JobClient> execute(@Nonnull final Pipeline pipeline, @Nonnull final Configuration configuration) throws Exception { 
    	// 获取JobGraph
    	final JobGraph jobGraph = PipelineExecutorUtils.getJobGraph(pipeline, configuration);
    	// 创建描述Yarn集群信息的ClusterDescriptor
    	try (final ClusterDescriptor<ClusterID> clusterDescriptor = clusterClientFactory.createClusterDescriptor(configuration)) { 
    		// 返回已经在运行的clusterID
    		final ClusterID clusterID = clusterClientFactory.getClusterId(configuration);
    		checkState(clusterID != null);
    		// 搜索得到该集群的Client
    		final ClusterClientProvider<ClusterID> clusterClientProvider = clusterDescriptor.retrieve(clusterID);
    		// RestClusterClient
    		ClusterClient<ClusterID> clusterClient = clusterClientProvider.getClusterClient();
    		// 往该集群提交jobGraph
    		return clusterClient
    				.submitJob(jobGraph)
    				.thenApplyAsync(jobID -> (JobClient) new ClusterClientJobClientAdapter<>(
    						clusterClientProvider,
    						jobID))
    				.whenComplete((ignored1, ignored2) -> clusterClient.close());
    	}
    }

### 4.3.2 submitJob ###

public CompletableFuture<JobID> submitJob(@Nonnull JobGraph jobGraph)

该方法利用`CompletableFuture`，异步执行以下过程：

1.  将jobGraph写入本地临时文件`flink-jobgraph`
2.  将flink-jobgraph、UserJars、本地UserArtifacts上传到HDFS
3.  成功后删除flink-jobgraph文件

也就是说，将JobGraph提交到了集群执行。

# 5 运行阶段-ExecutionGraph的生成 #

## 5.1 Local模式 ##

### 5.1.1 ExecutionGraph的生成 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 18]  
这里接的是之前的步骤5.2.4，即创建JobMaster的组件 DefaultSchduler的父类SchedulerBase时

this.executionGraph = createAndRestoreExecutionGraph(jobManagerJobMetricGroup, checkNotNull(shuffleMaster), checkNotNull(partitionTracker));

入口方法为`SchedulerBase#createAndRestoreExecutionGraph`，实际上会调用`ExecutionGraphBuilder.buildGraph`。主要步骤如下：

1.  创建包含Job基本信息的JobInformation  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 19]
2.  使用各项配置信息初始化一个`ExecutionGraph`  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 20]
3.  利用JobGraph生成JSON格式的执行计划
    
        { 
            "jid":"7721252eec9265198d189d4ea95f25be",
            "name":"Streaming WordCount",
            "nodes":[
                { 
                    "id":"001d41b9f8d4f77b8344bef265584b4b",
                    "parallelism":2,
                    "operator":"",
                    "operator_strategy":"",
                    "description":"aggregation -> Sink: Print to Std. Out",
                    "inputs":[
                        { 
                            "num":0,
                            "id":"de3774f5946826244eca962825ce5e73",
                            "ship_strategy":"HASH",
                            "exchange":"pipelined_bounded"
                        }
                    ],
                    "optimizer_properties":{ 
        
                    }
                },
                { 
                    "id":"de3774f5946826244eca962825ce5e73",
                    "parallelism":2,
                    "operator":"",
                    "operator_strategy":"",
                    "description":"Flat Map -> Filter -> Map",
                    "inputs":[
                        { 
                            "num":0,
                            "id":"bc764cd8ddf7a0cff126f51c16239658",
                            "ship_strategy":"REBALANCE",
                            "exchange":"pipelined_bounded"
                        }
                    ],
                    "optimizer_properties":{ 
        
                    }
                },
                { 
                    "id":"bc764cd8ddf7a0cff126f51c16239658",
                    "parallelism":1,
                    "operator":"",
                    "operator_strategy":"",
                    "description":"Source: Collection Source",
                    "optimizer_properties":{ 
        
                    }
                }
            ]
        }
4.  对JobVertex拓扑排序
5.  遍历已拓扑排序的JobVertex，生成`ExecutionJobVertex`  
    ExecutionJobVertex重要组件：
    
     *  ResourceProfile  
        描述不可变的Slot资源
     *  List inputs  
        上游的输入
     *  slotSharingGroup和coLocationGroup
     *  创建本节点生产的`IntermediateResult[] producedDataSets`的所有`IntermediateDataSet`元素  
        数组大小等于`jobVertex.getProducedDataSets().size()`即JobGraph中 该节点生产的IntermediateDataSet的数量。
        
            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());
            }
     *  ExecutionVertex\[parallelism\]  
        算子并行度大小的`ExecutionVertex`数组（比如我们例子中，Source就是1，FlatMap/Filter/Map就是2），初始化的时候就会创建所有ExecutionVertex。内部就包含了以下重要对象：
        
         *  ExecutionJobVertex
         *  subTaskIndex
         *  ExecutionVertexID  
            由JobVertexId和subTaskIndex得出
         *  taskNameWithSubtask  
            如`aggregation -> Sink: Print to Std. Out (1/2)`，小括号里表示`序号/算子总并发度`
         *  Map<IntermediateResultPartitionID, IntermediateResultPartition> resultPartitions
         *  根据本节点的`producedDataSets`大小，创建对应个数的`IntermediateResultPartition`，并记录到IntermediateResult的数组里
            
                for (IntermediateResult result : producedDataSets) { 
                	IntermediateResultPartition irp = new IntermediateResultPartition(result, this, subTaskIndex);
                	result.setPartition(subTaskIndex, irp);
                	resultPartitions.put(irp.getPartitionId(), irp);
                }
         *  创建本节点对应的JobVertex的input JobVertex数量 相等的`ExecutionEdge`二维数组，每个上游Input分支就是一个ExecutionEdge数组
         *  Execution  
            一个节点的单次执行。`ExecutionVertex`可以执行多次（恢复、重计算、重配置），而本类追踪该节点的一次执行及资源情况。
         *  CoLocationGroup限制\`locationConstraint
         *  InputSplit数组`inputSplits`
6.  对每个ExecutionJobVertex，会做以下连接处理：
    
    1.  将后继`ExecutionJobVertex`的输入`List<IntermediateResult> inputs`设为前驱结点的输出`IntermediateResult`；
    2.  遍历所有子ExecutionVertex， 调用`ExecutionGraphBuilder#connectSource`方法，根据该节点入边`JobEdge.distributionPattern`调用不同方法进行连接
    
     *  POINTWISE - connectPointwise
     *  ALL\_TO\_ALL - connectAllToAll  
        ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 21]  
        这种模式下，会将所有并行实例的`ExecutionVertex`创建一个`ExecutionEdge[sourcePartitions.length]`：
    
        ExecutionEdge[] edges = new ExecutionEdge[sourcePartitions.length];
        for (int i = 0; i < sourcePartitions.length; i++) { 
        	IntermediateResultPartition irp = sourcePartitions[i];
        	edges[i] = new ExecutionEdge(irp, this, inputNumber);
        }
        return edges;
    
    该ExecutionEdge的Source（IntermediateResultPartition）的Producer就是上游ExecutionVertex（如Flat Map -> Filter -> Map (1/2)），Consumer是这个数组的所有ExecutionEdge；**注意这里的Source数量和上游算子并行度相同。**
    
    而ExecutionEdgetarget是下游的ExecutionVertex，其`inputEdges`里就记录了上述的ExecutionEdge。
    
    因为是`ALL_TO_ALL`，所以需要为该算子其他ExecutionVertex实例也做上述操作，注意这里Source都是相同的，所以就是上游和下游所有实例节点都有了连接。
7.  开始配置State Checkpoint  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 22]
8.  初始化用户定义的Checkpoint Hook
9.  使用以上信息构建做Checkpoint服务相关初始化工作，如构建CheckpointCoordinator详见5.1.2
10. 返回构建好的ExecutionGraph

### 5.1.2 CheckpointCoordinator初始化 ###

### 5.1.3 尝试从Savepoint恢复ExecutionGraph ###

详见`SchedulerBase#tryRestoreExecutionGraphFromSavepoint`

### 5.1.4 创建好的ExecutionGraph ###

这里接的是之前的步骤5.2.4，即创建JobMaster的组件 DefaultSchduler的父类SchedulerBase时，

// 现在已经创建完了executionGraph
    this.executionGraph = createAndRestoreExecutionGraph(jobManagerJobMetricGroup, checkNotNull(shuffleMaster), checkNotNull(partitionTracker));
    // 执行拓扑
    this.schedulingTopology = executionGraph.getSchedulingTopology();

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 23]  
![在这里插入图片描述][20200908103253734.png_pic_center]

执行拓扑可以参照下面这张图，和我们的情况其实差不多  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 24]

*  ExecutionJobVertex个数等于JobVertex  
    这里就是Source、FlatMap->Filter->/Map、KeyedAggregaion->Sink三个节点
 *  根据并行度，创建对应个数的ExecutionVertex  
    比如上图中，因为Source并行度为1，所以只有一个ExecutionVertex；另外的节点并行度都为2，所以创建了两个ExecutionVertex。
 *  根据上下游输入输出关系，创建IntermediateResultPartition  
    IntermediateResultPartition是在从JobGraph阶段的IntermediateDataset的拓展出的IntermediateResult（多了Producer并行度信息和ExecutionJobVertex Producer信息）的基础之上，并根据Producer算子的并发量创建的。
    
    如这里FlatMap->Filter->/Map并发量为2，所以创建了两个ExecutionVertex和对应的两个`IntermediateResultPartition`。
    
    他包含了Producer ExecutionVertex、`List<List<ExecutionEdge>> consumers`等信息。
 *  根据JobGraph阶段的JobEdge、ExecutionGraph阶段上下游partition信息，创建ExecutionEdge  
    具体来说，是根据`DistributionPattern`不同来创建。比如我们这里是`ALL_TO_ALL`，则下游每个ExecutionVertex会为每个上游ExecutionVertex实例分别创建一个ExecutionEdge，并与上游的每个IntermediateResultPartition建立连接。
    
     *  该ExecutionEdge的Source（IntermediateResultPartition）的Producer就是上游ExecutionVertex（如Flat Map -> Filter -> Map (1/2)），Consumer是这个数组的所有ExecutionEdge；**注意这里的Source数量和上游算子并行度相同。**
     *  而ExecutionEdge的target是下游的ExecutionVertex，其`inputEdges`里就记录了上述的ExecutionEdge。

程序生成的执行拓扑`SchedulingTopology`如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 25]  
可以看到，这和我们前面提到的一些要点是相吻合的

*  37207和37208的是`FlatMap->Filter->Map` 的ExecutionVertex的两个实例，所以是各自对应IntermediateResultPartition的Producer
 *  37209和37210是`KeyedAggregation->Sink`的ExecutionVertex的两个实例，他们都是上游所有`IntermediateResultPartition`实例的Consumer
 *  pipelinedRegions只有一个，包括所有ExecutionVertex节点

### 5.1.5 启动JobMaster ###

见步骤6

### 5.1.6 运行阶段 ###

详见 第六章

## 5.2 Yarn Per-Job模式 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 26]

### 5.2.1 AM 入口类 YarnJobClusterEntrypoint ###

Flink在Yarn上作为AM跑的入口类为`YarnJobClusterEntrypoint`。

直接看其父类`ClusterEntrypoint`核心方法`runCluster`的核心代码

// 见4.2.5.2
    initializeServices(configuration, pluginManager);
    
    final DispatcherResourceManagerComponentFactory dispatcherResourceManagerComponentFactory = createDispatcherResourceManagerComponentFactory(configuration);
    
    // 该组件会在一个进程内启动Dispatcher、ResourceManager、WebMonitorEndpoint
    // 见4.2.5.3
    clusterComponent = dispatcherResourceManagerComponentFactory.create(
    	configuration,
    	ioExecutor,
    	commonRpcService,
    	haServices,
    	blobServer,
    	heartbeatServices,
    	metricRegistry,
    	archivedExecutionGraphStore,
    	new RpcMetricQueryServiceRetriever(metricRegistry.getMetricQueryServiceRpcService()),
    	this);

### 5.2.2 initializeServices ###

1.  创建`AkkaRpcService`类型的commonRpcService
2.  创建JM用来处理阻塞IO操作的`ioExecutor`线程池  
    核心线程默认为`Runtime.getRuntime().availableProcessors()`\*4，也可`Runtime.getRuntime().availableProcessors()`设置
3.  创建HA服务`HighAvailabilityServices`  
    ![在这里插入图片描述][20200902164804745.png_pic_center]
4.  创建并启动用于接收二进制大对象请求的`BlobServer`线程，是一个基于Socket的服务  
    监听相关请求，分配线程处理，创建目录存储BLOB文件或缓存他们
5.  创建用于心跳接收或请求的`HeartbeatServices`
6.  创建指标服务
7.  `ClusterEntrypoint#createSerializableExecutionGraphStore`，创建`MemoryArchivedExecutionGraphStore`  
    在内存存储`ArchivedExecutionGraph`，这个类表示序列化后的`ExecutionGraph`。
    
    Session模式下，使用的是`FileArchivedExecutionGraphStore`，会把归档后的ExecutionGraph写入磁盘，并在内存缓存最新使用的那个ExecutionGraph用来快速服务，并会定期清理存储。

### 5.2.3 dispatcherResourceManagerComponentFactory.create ###

在一个进程内启动Dispatcher、ResourceManager、WebMonitorEndpoint

## 5.3 Yarn Session模式 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 27]

注意此模式下，一个SessionCluster内多个Job共用一个Dispatcher和YarnResourceManager，而会为每个Job创建一个JobMaster来管理其JobGraph。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 28]

# 6 运行 #

## 6.1 Local模式 ##

### 6.1.1 JM侧 ###

#### 6.1.1.1 submitTask ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 29]

1.  JobMaster\#start  
    开启基于Akka的RPC服务，开始运行Job
2.  JobMaster\#startJobExecution
    
    1.  开启slotPool、scheduler，
    2.  通过Akka TCP 重连接到Flink ResourceManager，
    3.  `schedulerNG#startScheduling`  
        实际调用`SchedulerBase#startScheduling`，他会调用`DefaultScheduler#startSchedulingInternal`，先将`Execution.state`由`JobStatus.CREATED`转为`JobStatus.RUNNING`，然后调用`EagerSchedulingStrategy#allocateSlotsAndDeploy`开始为Job 的每个 ExecutionVertex申请Slot资源。最开始因为JobMaster尚未连接到ResourceManager，所以资源无法满足，`SlotPoolImpl`会打出如下日志：
        
            Cannot serve slot request, no ResourceManager connected. 
            Adding as pending request [SlotRequestId{ 37fa02beb31cd82b729577a18169af55}]
        
        此时需要创建`PendingRequest`，并把它放入`SlotPoolImpl.waitingForResourceManager`，等待分配Slot。
    4.  成功发现`ResourceManagerLeader`后，会回调`JobMaster#notifyOfNewResourceManagerLeader`，开始连接ResourceManager，成功后JobMaster通过动态代理的RPC来注册到ResourceManager，RM侧对应方法为`ResourceManager#registerJobManager`
    5.  （步骤2.5）JobMaster注册成功后，会触发监听机制，调用`slotPool#connectToResourceManager`  
        实际上是将前面`SlotPoolImpl.waitingForResourceManager`里的PendingRequest取出，并调用`slotPool#requestSlotFromResourceManager`来从ResourceManager申请Slot资源，会打出如下日志：
        
            Requesting new slot [SlotRequestId{ 37fa02beb31cd82b729577a18169af55}] 
            and profile ResourceProfile{ UNKNOWN} from resource manager.
        
        然后将pendingRequest记录到`pendingRequests`
    6.  (步骤2.6）`ResoureceManager#requestSlot`接收到SlotRequest请求后，会打出如下日志：
        
            Request slot with profile ResourceProfile{ UNKNOWN} for job 7fc0d98becdadc88693a5a43fa90ff6c with allocation id a65c076e1ca72384dda1b74cfd4c81b4.
        
        此时开始调用`slotManager.registerSlotRequest(slotRequest)`，SlotManagerImpl尝试为该请求分配Slot资源：
        
         *  如果`findMatchingSlot(ResourceProfile)`有可用`TaskManagerSlot`，就先从`freeSlots`中移除要分配的Slot，然后使用`SlotManagerImpl#allocateSlot`进行分配，此时就会利用`AkkaInvocationHandler`动态代理发送RPC（interface org.apache.flink.runtime.taskexecutor.TaskExecutorGateway\#requestSlot），而TM端对应响应详见6.1.2.2；
         *  否则调用`fulfillPendingSlotRequestWithPendingTaskManagerSlot`
3.  `DefaultSchedule#assignResourceOrHandleErrorr`
4.  `DefaultScheduler#deployTaskSafe`  
    在TM成功offerSlot给JM后（详见TM侧6.1.2.3、JM侧6.1.1.2），每个ExecutionVertex都会调用本方法，然后调用`ExecutionVertexOperations#deploy(executionVertex)`，最终走到`Execution#deploy`，将`ExecutionState`转为`DEPLOYING`，创建包含所有Task部署信息的`TaskDeploymentDescriptor`(ExectuionVetex):
    
        final TaskDeploymentDescriptor deployment = TaskDeploymentDescriptorFactory
        	.fromExecutionVertex(vertex, attemptNumber)
        	.createDeploymentDescriptor(
        		slot.getAllocationId(),
        		slot.getPhysicalSlotNumber(),
        		taskRestore,
        		producedPartitions.values());
    
    这一步JM会打出很多日志信息类似如下:
    
        GlobalGroupAggregate(groupBy=[date_str], select=[date_str, MAX(max$0) AS EXPR$0, COUNT(distinct$0 count$1) AS uv]) -> Calc(select=[EXPR$0, uv]) -> Sink: Sink(table=[hive.flink_meta.test_sink_uv_mysql_20200820], fields=[time_str, uv]) (2/12) (bd87df0aa6cb49784bb21ad923b5c82b) switched from SCHEDULED to DEPLOYING.
5.  最后调用`taskManagerGateway#submitTask`
6.  `RpcTaskManagerGateway#submitTask`  
    将Task部署描述信息`TaskDeploymentDescriptor`通过RPC发送给TaskExecutor，提交任务。
7.  TM侧的submitTask响应见6.1.2.4

#### 6.1.1.2 JobMaster\#offerSlots ####

TM注册到JM，并且JM是控制该Job的Leader时，TM会分配Slot给JM，详见6.1.2.3。

本方法是JobMaster收到TM offerSlots RPC请求后的响应，主要流程如下：

1.  从registeredTaskManagers通过taskManagerId匹配TM是否已注册
2.  slotPool\#offerSlots  
    将分配的所有`SlotOffer`做如下：
    
    1.  利用相关信息创建`AllocatedSlot`，该类表示JM从TM获得分配的一个Slot，表示TM分配资源的一个分片
    2.  从pendingRequests（放入的时机见 步骤2.5）中取出该`SlotOffer.AllocationId`对应的PendingRequest，只要匹配上了就将该AllocatedSlot和SlotRequestId记录到`allocatedSlots`，然后会触发`DefaultScheduler#assignResourceOrHandleError`里return的函数，核心代码：
        
            final ExecutionVertex executionVertex = getExecutionVertex(executionVertexId);
            final boolean sendScheduleOrUpdateConsumerMessage = deploymentHandle.getDeploymentOption().sendScheduleOrUpdateConsumerMessage();
            executionVertex
            	.getCurrentExecutionAttempt()
            	.registerProducedPartitions(logicalSlot.getTaskManagerLocation(), sendScheduleOrUpdateConsumerMessage);
            // 这里的logicalSlot就包含了JM的SlotRequestId、TM分配的AllocatedSlot
            executionVertex.tryAssignResource(logicalSlot);
    3.  `Execution#tryAssignResource(LogicalSlot)`，将Slot分配给ExecutionVertex.currentExecution
    4.  以上过程完成后，回到`SlotPoolImpl#offerSlot`方法，只要成功就返回true；否则就尝试满足其他PendingRequest
    5.  Scheduler开始deploy所有ExecutionVertex，详见6.1.1.1 步骤8

#### 6.1.1.3 JobMaster\#registerTaskManager ####

TM注册到JM时通过RPC调用本方法，详见6.1.2.2步骤7.

主要流程：

1.  建立与TM RPC连接
2.  将`ResourceID taskManagerId`注册到SlotPoolImpl
3.  将<taskManagerId, TM位置和连接信息>记录到registeredTaskManagers
4.  通过接收到的心跳来监控目标TM

### 6.1.2 TaskExecutor侧 ###

#### 6.1.2.1 概述 ####

TaskExecutor创建、Slot资源初始化、注册到JM详见4.1.3.2

TaskExecutor负责运行多个分配给他的Task，Task要点如下：

*  一个Task实例表示TM上的一个并行SubTask实例，内部嵌套了一个Flink算子，会消费输入数据、产生结果(IntermediateResultPartition)，并会和JM通信。
 *  Task对于其他Task无感知，也不知道重试信息，这些数据都由JM管理。
 *  Task只知道自己要运行的Code、配置以及要消费和生产的`IntermediateResultPartitionId`。
 *  每个Task由一个专门的线程负责运行

#### 6.1.2.2 requestSlot ####

步骤2.6中ResourceManager.SlotManager在为Slot请求匹配到可用`TaskManagerSlot`后，会通过RPC调用本方法，在TM端打出如下日志：

Receive slot request 8093771b2c1f0b598b9e25aee0c88d16 for job 7fc0d98becdadc88693a5a43fa90ff6c 
    from resource manager with leader id 00000000000000000000000000000000.

会调用`allocateSlot`方法分配Slot给JM：

1.  `taskSlotTable.isSlotFree(slotId.getSlotNumber()))`判断目标Slot是否空闲。  
    taskSlotTable已经在4.1.3.2步骤4中初始化Slot资源。
2.  如果目标Slot空闲，则`allocateSlot`开始分配Slot
    
    1.  先构建TaskSlot:`taskSlot = new TaskSlot<>(index, resourceProfile, memoryPageSize, jobId, allocationId);`，
    2.  然后<index, taskSlot>放入 taskSlots Map；
    3.  <allocationId, TaskSlot>放入 allocatedSlots Map；
    4.  从`slotsPerJob Map`中取出此jobId对应的`Set<AllocationID> slots`，将此次allocationId放入此set
    5.  成功会打出如下日志：
        
            Allocated slot for 8093771b2c1f0b598b9e25aee0c88d16.
    6.  如果分配失败，则抛出`SlotAllocationException`
3.  随后为该JM Leader创建`JobManagerLeaderListener`，并传入新建的`leaderRetrievalService`中start。  
    这里打出日志：
    
        Add job 7fc0d98becdadc88693a5a43fa90ff6c for job leader monitoring.
4.  为该Job创建`JobOrConnection`对象，记录到JobTable
5.  (步骤5)成功寻找到JM Leader后会回调`JobManagerLeaderListener#notifyLeaderAddress`，建立TM与JM的RPC连接，日志如下：
    
        Try to register at job manager akka.tcp://flink@192.168.1.1:11111/user/rpc/jobmanager_2 with leader id 00000000-0000-0000-0000-000000000000.
        
        Resolved JobManager address, beginning registration
6.  随后将TM注册到JM，成功后日志如下：
    
        Successful registration at job manager akka.tcp://flink@192.168.1.1:11111/user/rpc/jobmanager_2 for job 7fc0d98becdadc88693a5a43fa90ff6c.
7.  注册时，JM侧对应方法见 6.1.1.3
8.  如果TM已经和JM Leader建立连接(6.1.2.3)，则会调用`offerSlotsToJobManager`分配Slot给JM，详见6.1.2.3步骤3

#### 6.1.2.3 JobLeaderListenerImpl ####

1.  当一个JM获得某个Job的领导权，且可以建立到该JM的连接时，会调用本类`jobManagerGainedLeadership`方法，然后调用`TaskExecutor#establishJobManagerConnection`，打出如下日志：
    
        Establish JobManager connection for job 7fc0d98becdadc88693a5a43fa90ff6c.
2.  然后通过`jobManagerHeartbeatManager#monitorTarget`来通过心跳监控JM存活。
3.  使用`taskSlotTable.hasAllocatedSlots(jobId)`方法判断该Job是否拥有已分配的Slot资源（其实就是判断`slotsPerJob`中JobId对应的Set是否有元素），如果有会打出如下日志
    
        Offer reserved slots to the leader of job 7fc0d98becdadc88693a5a43fa90ff6c.
4.  jobMasterGateway\#offerSlots  
    将可分配的Slot通过RPC递交给JM ，JobMaster响应见6.1.1.2

#### 6.1.2.4 submitTask - 物理执行图生成 ####

TM侧运行Job流程如下

1.  `TaskExecutor#submitTask`
    
    1.  初始化TM各种服务，如`JobManagerTaskRestore、TaskStateManager、MemoryManager`
    2.  通过从TaskDeploymentDescriptor中获取的各种信息（如JobInformation、TaskInformation等）组装为`org.apache.flink.runtime.taskmanager.Task`。
    
        Task task = new Task(
        	jobInformation,
        	taskInformation,
        	tdd.getExecutionAttemptId(),
        	tdd.getAllocationId(),
        	tdd.getSubtaskIndex(),
        	tdd.getAttemptNumber(),
        	tdd.getProducedPartitions(),
        	tdd.getInputGates(),
        	tdd.getTargetSlotNumber(),
        	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());
    
     *  这里`tdd.getProducedPartitions()`被用来制作物理运行图阶段的`ResultPartition`，从ExecutionGraph阶段的IntermediateResult转来，还会为每个下游`inputChannel`(ExecutionEdge)生成一个`ResultSubPartition`实例`PipelinedSubpartition`
     *  `tdd.getInputGates()`被用来制作物理运行图阶段的`InputGate`，从ExecutionGraph阶段的ExecutionEdge转来。
        
        这里会为每个输入ExecutionEdge生成一个`InputChannel`实例，每个InputChannel都只从上游单个ResultSubPartition消费，所以每个InputChannel实例只包含一个`ResultPartitionID`。
        
        为该InputChannel创建一个实例级别的`BufferManager`，他被用来向SubTask级别的`Floating Buffers`（BufferPool）或自己独享的Buffer申请或归还内存空间。
        
        为`LocalRecoveredInputChannel`创建管理生产/消费的ResultPartition的`registeredPartitions Map`
        
        为`LocalRecoveredInputChannel`创建负责发布事件的`TaskEventDispatcher`
     *  ExecutionGraph阶段的ExecutionVertex也转变为了Task实例
     *  到这里，ExecutionGraph被成功转为物理执行图（虚拟概念，并没有这个实体对象），转变过程如下图：  
        ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 4]
    
    1.  组装完成后，TM打出该SubTask日志：
    
        Received task GlobalGroupAggregate(groupBy=[date_str], select=[date_str, MAX(max$0) AS EXPR$0, COUNT(distinct$0 count$1) AS uv]) -> Calc(select=[EXPR$0, uv]) -> Sink: Sink(table=[hive.flink_meta.test_sink_uv_mysql_20200820], fields=[time_str, uv]) (5/12).
    
    1.  从`taskSlotTable`匹配分配的Slot和本TM拥有的Slot，匹配上后启动`executingThread`线程，详见6.1.2.5
    2.  将使用的partition标记为`Bookkeeping`状态  
        具体调用`TaskExecutorPartitionTrackerImpl#startTrackingPartition(JobId, TaskExecutorPartitionInfo)`来监控、追踪所有部署的Partition

#### 6.1.2.5 Task主线程 ####

Task核心工作线程，主要流程如下：

1.  将`Task.ExecutionState`由`ExecutionState.CREATED`转为`ExecutionState.DEPLOYING`
2.  获取用户代码ClassLoader，可能会下载Job相关Jar/Class，依赖的重要类为`BlobLibraryCacheManager`，她会从`PermanentBlobService`下载相关lib jar，还会创建拥有这些类的引用的ClassLoader。打出日志类似：
    
        Loading JAR files for task Source: Collection Source (1/1) (e45b0684761a150c1457e2838d39f514) [DEPLOYING].
    
    这里使用的是`ChildFirstClassLoader`，会优先尝试自己从`URL`中加载类，如果不行再由ParentClassLoader加载类
3.  将Task注册到网络栈中，会打出日志如下：
    
        Registering task at network: Source: Collection Source (1/1) (e45b0684761a150c1457e2838d39f514) [DEPLOYING].
    
    为该Subtask的每个ResultPartition实例分别创建一个BufferPool实例，由该SubTask的所有ResultSubPartition共享，属于`Floating Buffers`  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 30]  
    随后将ResultPartition注册到`ResultPartitionManager.registeredPartitions`
    
    最后为`ResultPartitionID`创建对应的`TaskEventHandler`，用来发布和订阅，将它注册到`TaskEventDispatcher.registeredHandlers`
4.  初始化用户代码，创建`RuntimeEnvironment`（提供给执行代码访问如Task的name、并行度、Configuration、数据流Reader和Writer，以及TM提供的一系列组件如MemoryManager、IO Manager等），将executingThread.ContextClassLoader设为我们之前创建的`ChildFirstClassLoader`，然后调用`loadAndInstantiateInvokable`来把Task对应的目标类（如`org.apache.flink.streaming.runtime.tasks.SourceStreamTask`对应Source、`org.apache.flink.streaming.runtime.tasks.OneInputStreamTask`对应`aggregation -> Sink: Print to Std. Out (2/2)`）加载并实例化，放入`AbstractInvokable`
5.  开始Task核心工作
    
    1.  将`Task.ExecutionState`由`ExecutionState.DEPLOYING`转为`ExecutionState.RUNNING`
    2.  将task开始运行的情况通知TM和JM
    3.  调用我们已实例化的`AbstractInvokable#invoke`，这是核心方法，详见6.1.2.6
6.  执行完成后，释放所有资源

#### 6.1.2.6 StreamTask\#invoke ####

@Override
    public final void invoke() throws Exception { 
    	try { 
    		// 1.用算子信息创建operatorChain，如果是AbstractUdfStreamOperator算子会调用setup方法初始化
    		// 2.为OneInputStreamTask创建 StreamOneInputProcessor，他包括输入、输出、operatorChain
    		// 3.调用各个StreamOperator#initializeState方法，初始化operator相关状态
    		// 4.读取可能有的分区恢复状态信息
    		beforeInvoke();
    
    		// final check to exit early before starting to run
    		if (canceled) { 
    			throw new CancelTaskException();
    		}
    
    		// let the task do its work
    		// 主处理线程，见
    		runMailboxLoop();
    
    		// if this left the run() method cleanly despite the fact that this was canceled,
    		// make sure the "clean shutdown" is not attempted
    		if (canceled) { 
    			throw new CancelTaskException();
    		}
    		// 扫尾工作
    		afterInvoke();
    	}
    	catch (Exception invokeException) { 
    		try { 
    			cleanUpInvoke();
    		}
    		catch (Throwable cleanUpException) { 
    			throw (Exception) ExceptionUtils.firstOrSuppressed(cleanUpException, invokeException);
    		}
    		throw invokeException;
    	}
    	cleanUpInvoke();
    }

#### 6.1.2.7 StreamTask\#runMailboxLoop ####

主工作线程，调用栈：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 31]

# 7 总结 #

1.  用户提交Job时，先启动一个Client进程负责作业的编译与提交，将Job生成StreamNode，然后按一定规则（如OpertorChain等）转为JobGraph，然后提交到Flink Cluster
2.  如果是Yarn Session 模式，则AM在首次提交时就已经启动，此时Client直接将JobGraph提交给Dispatcher；
    
    而Yarn Per-Job模式下，需要现向Yarn RM申请Container资源启动AM，然后再将JobGraph提交给Dispatcher
3.  Dispatcher接收到Job的JobGraph后，为该Job启动一个JobMaster实例
4.  JobMaster为Job生成ExecutionGraph、运行时物理图，最后向Flink RM申请Slot资源
5.  首次申请时，Flink RM需要向Yarn RM申请Container资源启动Flink TM
6.  TM启动后主动连接Flink RM，并传送自己携带的Slot资源信息
7.  Flink RM收到TM注册的资源信息后，将分配信息告知JM
8.  JM将Job 运行时物理图部署到TM上运行

# 参考文档 #

*  [Apache Flink 进阶教程（七）：网络流控及反压剖析][Apache Flink]
 *  [Flink 原理与实现：理解 Flink 中的计算资源][Flink _ Flink]
 *  《Stream Processing with Apache Flink》
    
     *  作者：崔兴灿
     *  出处：阿里巴巴-Flink极客训练营
 *  《Flink Runtime Architecture》
    
     *  作者：朱翥
     *  出处：阿里巴巴-Flink极客训练营
 *  [Apache Flink 进阶教程（一）：Runtime 核心机制剖析][Apache Flink _Runtime]
    
     *  作者：高赟
     *  出处：Ververica

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/9d7ee99e12864c30b9236f6805765166.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221123/2c5405b7e875402787e39f27797b2b43.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221123/2019341094c54cc2bfb17b08a4a17472.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221123/febf94f157da4545aefaa735f1775b26.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 4]: /images/20221123/a8a20863ea254d82b715ffe4b5de72e8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 5]: /images/20221123/02a06a8bcd8543288563ff31eb098218.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 6]: /images/20221123/4f7106d8072f4905a37942ce13fa6701.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 7]: /images/20221123/59fc31a3445a42ed911b3d90ca9e0fef.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 8]: /images/20221123/c9c5dc68f291405f90e08acd37d1cbae.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 9]: /images/20221123/790c2cd212894dbeadd1aa42a7152629.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 10]: /images/20221123/d50f144e49684cbc973af264bdbee282.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 11]: /images/20221123/6c8cb9465a934d21b0a499aab3792187.png
[20200902164804745.png_pic_center]: /images/20221123/e3ce0f45d7834027b4a6ea20c5d18456.png
[20200916112040650.png_pic_center]: /images/20221123/0cd3ed31b6cd4d1a9bc371204ee27c45.png
[20200916112421794.png_pic_center]: /images/20221123/36842258db534b57b1cc71d51b23d160.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 12]: /images/20221123/f568661135ba489ea9a3950c25bf1297.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 13]: /images/20221123/dcafe886202243b694f0e7ef93c076c1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 14]: /images/20221123/704bebc944404e5eac114d272ce1d193.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 15]: /images/20221123/b3f2fb0f1dd34e8990bb8e2dd97a4ff2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 16]: /images/20221123/aa67ecbb71a54b41b96860611f62d888.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 17]: /images/20221123/883d4131a6ef401cb8a5dfdd4b14f337.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 18]: /images/20221123/1963a781ef5646d5bd582536fe8af8ce.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 19]: /images/20221123/c0ff5cf3e0e44151a8b5702e3e4a74e7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 20]: /images/20221123/ef3ac8617b09447ca63759ad11fc38c4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 21]: /images/20221123/6e0cd86bff4b4c5487ee02589b493da0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 22]: /images/20221123/54adcee9236241de80d9944541d59963.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 23]: /images/20221123/06d5c52fabd64d4cac37330d1176bb23.png
[20200908103253734.png_pic_center]: /images/20221123/8a8e1cb0c2f345499424a92f5fcf2f53.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 24]: /images/20221123/d0a8f1b53f45410f8f004ebb39a9082e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 25]: /images/20221123/e54c126717c54d42aab91e6310e51d8d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 26]: /images/20221123/70c1736daa3b4ce0b8a9d089445b16f8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 27]: /images/20221123/92bebc573e35409d9cee9504f4f39e4a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 28]: /images/20221123/02730e0a2b754fc68508df596059ca76.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 29]: /images/20221123/418c832f98ca4cffaf802c789efd1eec.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 30]: /images/20221123/67cd7c3be5ad46a59b1c1fb43ed3c432.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 31]: /images/20221123/4feba6aeb69f4dcab6e563461698903f.png
[Apache Flink]: https://ververica.cn/developers/advanced-tutorial-2-analysis-of-network-flow-control-and-back-pressure/
[Flink _ Flink]: http://wuchong.me/blog/2016/05/09/flink-internals-understanding-execution-resources/
[Apache Flink _Runtime]: https://ververica.cn/developers/advanced-tutorial-1-analysis-of-the-core-mechanism-of-runtime/