Flink 源码入门02

Dear 丶 2021-12-03 03:41 316阅读 0赞

事实上，flink总共提供了三种图的抽象，我们前面已经提到了StreamGraph和JobGraph，还有一种是ExecutionGraph，是用于调度的基本数据结构。   
![image\_1caf1oll019fp1odv1bh9idosr79.png-486.3kB][image_1caf1oll019fp1odv1bh9idosr79.png-486.3kB]

上面这张图清晰的给出了flink各个图的工作原理和转换过程。其中最后一个物理执行图并非flink的数据结构，而是程序开始执行后，各个task分布在不同的节点上，所形成的物理上的关系表示。

*  从JobGraph的图里可以看到，数据从上一个operator流到下一个operator的过程中，上游作为生产者提供了IntermediateDataSet，而下游作为消费者需要JobEdge。事实上，JobEdge是一个通信管道，连接了上游生产的dataset和下游的JobVertex节点。
 *  在JobGraph转换到ExecutionGraph的过程中，主要发生了以下转变： 
    
     *  加入了并行度的概念，成为真正可调度的图结构
     *  生成了与JobVertex对应的ExecutionJobVertex，ExecutionVertex，与IntermediateDataSet对应的IntermediateResult和IntermediateResultPartition等，并行将通过这些类实现
 *  ExecutionGraph已经可以用于调度任务。我们可以看到，flink根据该图生成了一一对应的Task，每个task对应一个ExecutionGraph的一个Execution。Task用InputGate、InputChannel和ResultPartition对应了上面图中的IntermediateResult和ExecutionEdge。

那么，flink抽象出这三层图结构，四层执行逻辑的意义是什么呢？   
StreamGraph是对用户逻辑的映射。JobGraph在此基础上进行了一些优化，比如把一部分操作串成chain以提高效率。ExecutionGraph是为了调度存在的，加入了并行处理的概念。而在此基础上真正执行的是Task及其相关结构。

在第一节的算子注册部分，我们可以看到，flink把每一个算子transform成一个对流的转换（比如上文中返回的SingleOutputStreamOperator是一个DataStream的子类），并且注册到执行环境中，用于生成StreamGraph。实际生成StreamGraph的入口是`StreamGraphGenerator.generate(env, transformations)` 其中的transformations是一个list，里面记录的就是我们在transform方法中放进来的算子。

![2019072209225395.png][]

创建StreamGrraph：

![20190722092347166.png][]

**StreamTransformation 类，代表流算子转换**

StreamTransformation代表了从一个或多个DataStream生成新DataStream的操作。顺便，DataStream类在内部组合了一个StreamTransformation类，实际的转换操作均通过该类完成 。

它的构造器：

package org.apache.flink.streaming.api.transformations;

![20190722092522663.png][]

![image\_1caf64b7c1gjnv2eebi1v9e1cvum.png-129.4kB][image_1caf64b7c1gjnv2eebi1v9e1cvum.png-129.4kB]

我们可以看到，从source到各种map,union再到sink操作全部被映射成了StreamTransformation。   
其映射过程如下所示：

![image\_1caf6ak4rkqsc1u1hci93fe0d13.png-36.6kB][image_1caf6ak4rkqsc1u1hci93fe0d13.png-36.6kB]

以MapFunction为例：

*  首先，用户代码里定义的UDF会被当作其基类对待，然后交给StreamMap这个operator做进一步包装。事实上，每一个Transformation都对应了一个StreamOperator。
 *  由于map这个操作只接受一个输入，所以再被进一步包装为OneInputTransformation。
 *  最后，将该transformation注册到执行环境中，当执行上文提到的generate方法时，生成StreamGraph图结构。
    
    > 另外，并不是每一个 StreamTransformation 都会转换成runtime层中的物理操作。有一些只是逻辑概念，比如union、split/select、partition等。如下图所示的转换树，在运行时会优化成下方的操作图。   
    > ![image\_1caf71h79s0s3fodem1aeb1j3m1g.png-83.8kB][image_1caf71h79s0s3fodem1aeb1j3m1g.png-83.8kB]

StreamGraph生成函数分析 ：

我们从StreamGraphGenerator.generate()方法往下看：

package org.apache.flink.streaming.api.graph;

> public static StreamGraph generate(StreamExecutionEnvironment env, List<StreamTransformation<?>> transformations) \{  
>         return new StreamGraphGenerator(env).generateInternal(transformations);  
>     \}  
>       
>     //注意，StreamGraph的生成是从sink开始的  
>     private StreamGraph generateInternal(List<StreamTransformation<?>> transformations) \{  
>         for (StreamTransformation<?> transformation: transformations) \{  
>             transform(transformation);  
>         \}  
>         return streamGraph;  
>     \}  
>       
>     //这个方法的核心逻辑就是判断传入的steamOperator是哪种类型，并执行相应的操作，详情见下面那一大堆if-else  
>     private Collection<Integer> transform(StreamTransformation<?> transform) \{
> 
>         if (alreadyTransformed.containsKey(transform)) \{  
>             return alreadyTransformed.get(transform);  
>         \}
> 
>         LOG.debug("Transforming " + transform);
> 
>         if (transform.getMaxParallelism() <= 0) \{
> 
>             // if the max parallelism hasn't been set, then first use the job wide max parallelism  
>             // from theExecutionConfig.  
>             int globalMaxParallelismFromConfig = env.getConfig().getMaxParallelism();  
>             if (globalMaxParallelismFromConfig > 0) \{  
>                 transform.setMaxParallelism(globalMaxParallelismFromConfig);  
>             \}  
>         \}
> 
>         // call at least once to trigger exceptions about MissingTypeInfo  
>         transform.getOutputType();
> 
>         Collection<Integer> transformedIds;  
>         //这里对操作符的类型进行判断，并以此调用相应的处理逻辑.简而言之，处理的核心无非是递归的将该节点和节点的上游节点加入图  
>         if (transform instanceof OneInputTransformation<?, ?>) \{  
>             transformedIds = transformOneInputTransform((OneInputTransformation<?, ?>) transform);  
>         \} else if (transform instanceof TwoInputTransformation<?, ?, ?>) \{  
>             transformedIds = transformTwoInputTransform((TwoInputTransformation<?, ?, ?>) transform);  
>         \} else if (transform instanceof SourceTransformation<?>) \{  
>             transformedIds = transformSource((SourceTransformation<?>) transform);  
>         \} else if (transform instanceof SinkTransformation<?>) \{  
>             transformedIds = transformSink((SinkTransformation<?>) transform);  
>         \} else if (transform instanceof UnionTransformation<?>) \{  
>             transformedIds = transformUnion((UnionTransformation<?>) transform);  
>         \} else if (transform instanceof SplitTransformation<?>) \{  
>             transformedIds = transformSplit((SplitTransformation<?>) transform);  
>         \} else if (transform instanceof SelectTransformation<?>) \{  
>             transformedIds = transformSelect((SelectTransformation<?>) transform);  
>         \} else if (transform instanceof FeedbackTransformation<?>) \{  
>             transformedIds = transformFeedback((FeedbackTransformation<?>) transform);  
>         \} else if (transform instanceof CoFeedbackTransformation<?>) \{  
>             transformedIds = transformCoFeedback((CoFeedbackTransformation<?>) transform);  
>         \} else if (transform instanceof PartitionTransformation<?>) \{  
>             transformedIds = transformPartition((PartitionTransformation<?>) transform);  
>         \} else if (transform instanceof SideOutputTransformation<?>) \{  
>             transformedIds = transformSideOutput((SideOutputTransformation<?>) transform);  
>         \} else \{  
>             throw new IllegalStateException("Unknown transformation: " + transform);  
>         \}
> 
>         //注意这里和函数开始时的方法相对应，在有向图中要注意避免循环的产生  
>         // need this check because the iterate transformation adds itself before  
>         // transforming the feedback edges  
>         if (!alreadyTransformed.containsKey(transform)) \{  
>             alreadyTransformed.put(transform, transformedIds);  
>         \}
> 
>         if (transform.getBufferTimeout() > 0) \{  
>             streamGraph.setBufferTimeout(transform.getId(), transform.getBufferTimeout());  
>         \}  
>         if (transform.getUid() != null) \{  
>             streamGraph.setTransformationUID(transform.getId(), transform.getUid());  
>         \}  
>         if (transform.getUserProvidedNodeHash() != null) \{  
>             streamGraph.setTransformationUserHash(transform.getId(), transform.getUserProvidedNodeHash());  
>         \}
> 
>         if (transform.getMinResources() != null && transform.getPreferredResources() != null) \{  
>             streamGraph.setResources(transform.getId(), transform.getMinResources(), transform.getPreferredResources());  
>         \}
> 
>         return transformedIds;  
>     \}

已经有的转化算子：

![20190722093136115.png][]

因为map，filter等常用操作都是OneInputStreamOperator,我们就来看看`transformOneInputTransform((OneInputTransformation<?, ?>) transform)`方法。

package org.apache.flink.streaming.api.graph;

> private <IN, OUT> Collection<Integer> transformOneInputTransform(OneInputTransformation<IN, OUT> transform) \{
> 
>         Collection<Integer> inputIds = transform(transform.getInput());
> 
>         // 在递归处理节点过程中，某个节点可能已经被其他子节点先处理过了，需要跳过  
>         if (alreadyTransformed.containsKey(transform)) \{  
>             return alreadyTransformed.get(transform);  
>         \}
> 
>         //这里是获取slotSharingGroup。这个group用来定义当前我们在处理的这个操作符可以跟什么操作符chain到一个slot里进行操作  
>         //因为有时候我们可能不满意flink替我们做的chain聚合  
>         //一个slot就是一个执行task的基本容器  
>         String slotSharingGroup = determineSlotSharingGroup(transform.getSlotSharingGroup(), inputIds);
> 
>         //把该operator加入图  
>         streamGraph.addOperator(transform.getId(),  
>                 slotSharingGroup,  
>                 transform.getOperator(),  
>                 transform.getInputType(),  
>                 transform.getOutputType(),  
>                 transform.getName());  
>           
>         //对于keyedStream，我们还要记录它的keySelector方法  
>         //flink并不真正为每个keyedStream保存一个key，而是每次需要用到key的时候都使用keySelector方法进行计算  
>         //因此，我们自定义的keySelector方法需要保证幂等性  
>         //到后面介绍keyGroup的时候我们还会再次提到这一点  
>         if (transform.getStateKeySelector() != null) \{  
>             TypeSerializer<?> keySerializer = transform.getStateKeyType().createSerializer(env.getConfig());  
>             streamGraph.setOneInputStateKey(transform.getId(), transform.getStateKeySelector(), keySerializer);  
>         \}
> 
>         streamGraph.setParallelism(transform.getId(), transform.getParallelism());  
>         streamGraph.setMaxParallelism(transform.getId(), transform.getMaxParallelism());  
>           
>         //为当前节点和它的依赖节点建立边  
>         //这里可以看到之前提到的select union partition等逻辑节点被合并入edge的过程  
>         for (Integer inputId: inputIds) \{  
>             streamGraph.addEdge(inputId, transform.getId(), 0);  
>         \}
> 
>         return Collections.singleton(transform.getId());  
>     \}  
>       
>     public void addEdge(Integer upStreamVertexID, Integer downStreamVertexID, int typeNumber) \{  
>         addEdgeInternal(upStreamVertexID,  
>                 downStreamVertexID,  
>                 typeNumber,  
>                 null,  
>                 new ArrayList<String>(),  
>                 null);
> 
>     \}  
>     //addEdge的实现，会合并一些逻辑节点  
>     private void addEdgeInternal(Integer upStreamVertexID,  
>             Integer downStreamVertexID,  
>             int typeNumber,  
>             StreamPartitioner<?> partitioner,  
>             List<String> outputNames,  
>             OutputTag outputTag) \{  
>         //如果输入边是侧输出节点，则把side的输入边作为本节点的输入边，并递归调用  
>         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);  
>             //如果输入边是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);  
>             //如果是partition节点  
>         \} else if (virtualPartitionNodes.containsKey(upStreamVertexID)) \{  
>             int virtualId = upStreamVertexID;  
>             upStreamVertexID = virtualPartitionNodes.get(virtualId).f0;  
>             if (partitioner == null) \{  
>                 partitioner = virtualPartitionNodes.get(virtualId).f1;  
>             \}  
>             addEdgeInternal(upStreamVertexID, downStreamVertexID, typeNumber, partitioner, outputNames, outputTag);  
>         \} else \{  
>         //正常的edge处理逻辑  
>             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.  
>             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.");  
>                 \}  
>             \}
> 
>             StreamEdge edge = new StreamEdge(upstreamNode, downstreamNode, typeNumber, outputNames, partitioner, outputTag);
> 
>             getStreamNode(edge.getSourceId()).addOutEdge(edge);  
>             getStreamNode(edge.getTargetId()).addInEdge(edge);  
>         \}  
>     \}

WordCount函数的StreamGraph：

flink提供了一个StreamGraph可视化显示工具，[在这里][Link 1]   
我们可以把我们的程序的执行计划打印出来`System.out.println(env.getExecutionPlan());` 复制到这个网站上，点击生成，如图所示：   
![image\_1cafgsliu1n2n1uj21p971b0h6m71t.png-25.7kB][image_1cafgsliu1n2n1uj21p971b0h6m71t.png-25.7kB]   
可以看到，我们源程序被转化成了4个operator。   
另外，在operator之间的连线上也显示出了flink添加的一些逻辑流程。由于我设定了每个操作符的并行度都是1，所以在每个操作符之间都是直接FORWARD，不存在shuffle的过程。

### 2.3 JobGraph的生成 ###

flink会根据上一步生成的StreamGraph生成JobGraph，然后将JobGraph发送到server端进行ExecutionGraph的解析。

**2.3.1 JobGraph生成源码**

与StreamGraph类似，JobGraph的入口方法是`StreamingJobGraphGenerator.createJobGraph()`。我们直接来看源码

package org.apache.flink.streaming.api.graph;

>     private JobGraph createJobGraph() {
>     
>             // 设置启动模式为所有节点均在一开始就启动
>             jobGraph.setScheduleMode(ScheduleMode.EAGER);
>     
>             // 为每个节点生成hash id
>             Map<Integer, byte[]> hashes = defaultStreamGraphHasher.traverseStreamGraphAndGenerateHashes(streamGraph);
>     
>             // 为了保持兼容性创建的hash
>             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<>();
>             //生成jobvertex，串成chain等
>             //这里的逻辑大致可以理解为，挨个遍历节点，如果该节点是一个chain的头节点，就生成一个JobVertex，如果不是头节点，就要把自身配置并入头节点，然后把头节点和自己的出边相连；对于不能chain的节点，当作只有头节点处理即可
>             setChaining(hashes, legacyHashes, chainedOperatorHashes);
>             //设置输入边edge
>             setPhysicalEdges();
>             //设置slot共享group
>             setSlotSharing();
>             //配置检查点
>             configureCheckpointing();
>     
>             // 如果有之前的缓存文件的配置的话，重新读入
>             for (Tuple2<String, DistributedCache.DistributedCacheEntry> e : streamGraph.getEnvironment().getCachedFiles()) {
>                 DistributedCache.writeFileInfoToConfig(e.f0, e.f1, jobGraph.getJobConfiguration());
>             }
>     
>             // 传递执行环境配置
>             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;
>         }

**2.3.2 operator chain的逻辑**

为了更高效地分布式执行，Flink会尽可能地将operator的subtask链接（chain）在一起形成task。每个task在一个线程中执行。将operators链接成task是非常有效的优化：它能减少线程之间的切换，减少消息的序列化/反序列化，减少数据在缓冲区的交换，减少了延迟的同时提高整体的吞吐量。

![image\_1cafj7s6bittk5tt0bequlig2a.png-158.7kB][image_1cafj7s6bittk5tt0bequlig2a.png-158.7kB]

上图中将KeyAggregation和Sink两个operator进行了合并，因为这两个合并后并不会改变整体的拓扑结构。但是，并不是任意两个 operator 就能 chain 一起的,其条件还是很苛刻的：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70][]

flink的chain逻辑是一种很常见的设计，比如spring的interceptor也是类似的实现方式。通过把操作符串成一个大操作符，flink避免了把数据序列化后通过网络发送给其他节点的开销，能够大大增强效率

**2.3.3 JobGraph的提交**

前面已经提到，JobGraph的提交依赖于JobClient和JobManager之间的异步通信，如图所示：   
![image\_1cafn516r1p68kt31g7r196rcsv2n.png-40.1kB][image_1cafn516r1p68kt31g7r196rcsv2n.png-40.1kB]

在submitJobAndWait方法中，其首先会创建一个JobClientActor的ActorRef,然后向其发起一个SubmitJobAndWait消息，该消息将JobGraph的实例提交给JobClientActor。发起模式是ask，它表示需要一个应答消息。

**（这部分代码没有在源码里面找到，可能是版本变化了，这部分代码不在了 ）**

Future<Object> future = Patterns.ask(jobClientActor, new JobClientMessages.SubmitJobAndWait(jobGraph), new Timeout(AkkaUtils.INF_TIMEOUT()));
    answer = Await.result(future, AkkaUtils.INF_TIMEOUT());

该SubmitJobAndWait消息被JobClientActor接收后，最终通过调用tryToSubmitJob方法触发真正的提交动作。当JobManager的actor接收到来自client端的请求后，会执行一个submitJob方法，主要做以下事情：

>  *  向BlobLibraryCacheManager注册该Job；
>  *  构建ExecutionGraph对象；
>  *  对JobGraph中的每个顶点进行初始化；
>  *  将DAG拓扑中从source开始排序，排序后的顶点集合附加到Exec> - utionGraph对象；
>  *  获取检查点相关的配置，并将其设置到ExecutionGraph对象；
>  *  向ExecutionGraph注册相关的listener；
>  *  执行恢复操作或者将JobGraph信息写入SubmittedJobGraphStore以在后续用于恢复目的；
>  *  响应给客户端JobSubmitSuccess消息；
>  *  对ExecutionGraph对象进行调度执行；

最后，JobManger会返回消息给JobClient，通知该任务是否提交成功。

### 2.4 ExecutionGraph的生成 ###

与StreamGraph和JobGraph不同，ExecutionGraph并不是在我们的客户端程序生成，而是**在服务端（JobManager处）生成的**，顺便flink只维护一个JobManager。其入口代码是：

package org.apache.flink.runtime.executiongraph;

`ExecutionGraphBuilder.buildGraph（...）`   该方法长200多行，其中一大半是checkpoiont的相关逻辑，我们暂且略过，直接看核心方法`executionGraph.attachJobGraph(sortedTopology)`   
因为ExecutionGraph事实上只是改动了JobGraph的每个节点，而没有对整个拓扑结构进行变动，所以代码里只是挨个遍历jobVertex并进行处理：

package org.apache.flink.runtime.executiongraph;

> for (JobVertex jobVertex : topologiallySorted) \{
> 
>             if (jobVertex.isInputVertex() && !jobVertex.isStoppable()) \{  
>                 this.isStoppable = false;  
>             \}
> 
>             //在这里生成ExecutionGraph的每个节点  
>             //首先是进行了一堆赋值，将任务信息交给要生成的图节点，以及设定并行度等等  
>             //然后是创建本节点的IntermediateResult，根据本节点的下游节点的个数确定创建几份  
>             //最后是根据设定好的并行度创建用于执行task的ExecutionVertex  
>             //如果job有设定inputsplit的话，这里还要指定inputsplits  
>             ExecutionJobVertex ejv = new ExecutionJobVertex(  
>                 this,  
>                 jobVertex,  
>                 1,  
>                 rpcCallTimeout,  
>                 globalModVersion,  
>                 createTimestamp);  
>               
>             //这里要处理所有的JobEdge  
>             //对每个edge，获取对应的intermediateResult，并记录到本节点的输入上  
>             //最后，把每个ExecutorVertex和对应的IntermediateResult关联起来  
>             ejv.connectToPredecessors(this.intermediateResults);
> 
>             ExecutionJobVertex previousTask = this.tasks.putIfAbsent(jobVertex.getID(), ejv);  
>             if (previousTask != null) \{  
>                 throw new JobException(String.format("Encountered two job vertices with ID %s : previous=\[%s\] / new=\[%s\]",  
>                         jobVertex.getID(), ejv, previousTask));  
>             \}
> 
>             for (IntermediateResult res : ejv.getProducedDataSets()) \{  
>                 IntermediateResult previousDataSet = this.intermediateResults.putIfAbsent(res.getId(), res);  
>                 if (previousDataSet != null) \{  
>                     throw new JobException(String.format("Encountered two intermediate data set with ID %s : previous=\[%s\] / new=\[%s\]",  
>                             res.getId(), res, previousDataSet));  
>                 \}  
>             \}
> 
>             this.verticesInCreationOrder.add(ejv);  
>             this.numVerticesTotal += ejv.getParallelism();  
>             newExecJobVertices.add(ejv);  
>         \}

至此，ExecutorGraph就创建完成了。

## 3. 任务的调度与执行 ##

关于flink的任务执行架构，官网的这两张图就是最好的说明：   
![image\_1cafnu1pl1d8c15m219b8vkb2334.png-112.9kB][image_1cafnu1pl1d8c15m219b8vkb2334.png-112.9kB]   
Flink 集群启动后，首先会启动一个 JobManger 和多个的 TaskManager。用户的代码会由JobClient 提交给 JobManager，JobManager 再把来自不同用户的任务发给 不同的TaskManager 去执行，每个TaskManager管理着多个task，task是执行计算的最小结构， TaskManager 将心跳和统计信息汇报给 JobManager。TaskManager 之间以流的形式进行数据的传输。上述除了task外的三者均为独立的 JVM 进程。   
要注意的是，TaskManager和job并非一一对应的关系。flink调度的最小单元是task而非TaskManager，也就是说，来自不同job的不同task可能运行于同一个TaskManager的不同线程上。   
![image\_1cclle7ui2j41nf611gs1is18m19.png-127.5kB][image_1cclle7ui2j41nf611gs1is18m19.png-127.5kB]   
一个flink任务所有可能的状态如上图所示。图上画的很明白，就不再赘述了。

### 3.1 计算资源的调度 ###

**Task slot是一个TaskManager内资源分配的最小载体，代表了一个固定大小的资源子集**，每个TaskManager会将其所占有的资源平分给它的slot。   
通过调整 task slot 的数量，用户可以定义task之间是如何相互隔离的。**每个 TaskManager 有一个slot，也就意味着每个task运行在独立的 JVM 中。每个 TaskManager 有多个slot的话，也就是说多个task运行在同一个JVM中**。   
而在同一个JVM进程中的task，可以共享TCP连接（基于多路复用）和心跳消息，可以减少数据的网络传输，也能共享一些数据结构，一定程度上减少了每个task的消耗。

每个slot可以接受单个task，也可以接受多个连续task组成的pipeline，如下图所示，FlatMap函数占用一个taskslot，而key Agg函数和sink函数共用一个taskslot：

![image\_1cafpf21c1jh3s5ap1fisu4v23h.png-44.7kB][image_1cafpf21c1jh3s5ap1fisu4v23h.png-44.7kB]

为了达到共用slot的目的，除了可以以chain的方式pipeline算子，我们还可以允许SlotSharingGroup，如下图所示：   
![image\_1cafpko68b3r1lk0dpsnmbj3c3u.png-61.2kB][image_1cafpko68b3r1lk0dpsnmbj3c3u.png-61.2kB]

我们可以把不能被chain成一条的两个操作如flatmap和key&sink放在一个TaskSlot里执行，这样做可以获得以下好处：

*  共用slot使得我们不再需要计算每个任务需要的总task数目，直接取最高算子的并行度即可
 *  对计算资源的利用率更高。例如，通常的轻量级操作map和重量级操作Aggregate不再分别需要一个线程，而是可以在同一个线程内执行，而且对于slot有限的场景，我们可以增大每个task的并行度了。   
    接下来我们还是用官网的图来说明flink是如何重用slot的： ![image\_1cafqroarkjkuje1hfi18gor654b.png-137kB][image_1cafqroarkjkuje1hfi18gor654b.png-137kB]
    
    1.  TaskManager1分配一个SharedSlot0
    2.  把source task放入一个SimpleSlot0，再把该slot放入SharedSlot0
    3.  把flatmap task放入一个SimpleSlot1，再把该slot放入SharedSlot0
    4.  因为我们的flatmap task并行度是2，因此不能再放入SharedSlot0，所以向TaskMange21申请了一个新的SharedSlot0
    5.  把第二个flatmap task放进一个新的SimpleSlot，并放进TaskManager2的SharedSlot0
    6.  开始处理key&sink task，因为其并行度也是2，所以先把第一个task放进TaskManager1的SharedSlot
    7.  把第二个key&sink放进TaskManager2的SharedSlot

### 3.2 JobManager执行job ###

JobManager负责接收 flink 的作业，调度 task，收集 job 的状态、管理 TaskManagers。被实现为一个 akka actor。

3.2.1 JobManager的组件

*  BlobServer 是一个用来管理二进制大文件的服务，比如保存用户上传的jar文件，该服务会将其写到磁盘上。还有一些相关的类，如BlobCache，用于TaskManager向JobManager下载用户的jar文件 
    
        package org.apache.flink.runtime.blob;
 *  InstanceManager 用来管理当前存活的TaskManager的组件，记录了TaskManager的心跳信息等
    
        package org.apache.flink.runtime.instance;
 *  CompletedCheckpointStore 用于保存已完成的checkpoint相关信息，持久化到内存中或者zookeeper上
    
        package org.apache.flink.runtime.checkpoint;
 *  MemoryArchivist 保存了已经提交到flink的作业的相关信息，如JobGraph等

**3.2.2 JobManager的启动过程**

先列出JobManager启动的核心代码：

> def runJobManager(  
>       configuration: Configuration,  
>       executionMode: JobManagerMode,  
>       listeningAddress: String,  
>       listeningPort: Int)  
>     : Unit = \{
> 
>     val numberProcessors = Hardware.getNumberCPUCores()
> 
>     val futureExecutor = Executors.newScheduledThreadPool(  
>       numberProcessors,  
>       new ExecutorThreadFactory("jobmanager-future"))
> 
>     val ioExecutor = Executors.newFixedThreadPool(  
>       numberProcessors,  
>       new ExecutorThreadFactory("jobmanager-io"))
> 
>     val timeout = AkkaUtils.getTimeout(configuration)
> 
>     // we have to first start the JobManager ActorSystem because this determines the port if 0  
>     // was chosen before. The method startActorSystem will update the configuration correspondingly.  
>     val jobManagerSystem = startActorSystem(  
>       configuration,  
>       listeningAddress,  
>       listeningPort)
> 
>     val highAvailabilityServices = HighAvailabilityServicesUtils.createHighAvailabilityServices(  
>       configuration,  
>       ioExecutor,  
>       AddressResolution.NO\_ADDRESS\_RESOLUTION)
> 
>     val metricRegistry = new MetricRegistryImpl(  
>       MetricRegistryConfiguration.fromConfiguration(configuration))
> 
>     metricRegistry.startQueryService(jobManagerSystem, null)
> 
>     val (\_, \_, webMonitorOption, \_) = try \{  
>       startJobManagerActors(  
>         jobManagerSystem,  
>         configuration,  
>         executionMode,  
>         listeningAddress,  
>         futureExecutor,  
>         ioExecutor,  
>         highAvailabilityServices,  
>         metricRegistry,  
>         classOf\[JobManager\],  
>         classOf\[MemoryArchivist\],  
>         Option(classOf\[StandaloneResourceManager\])  
>       )  
>     \} catch \{  
>       case t: Throwable =>  
>         futureExecutor.shutdownNow()  
>         ioExecutor.shutdownNow()
> 
>         throw t  
>     \}
> 
>     // block until everything is shut down  
>     jobManagerSystem.awaitTermination()  
>       
>     .......  
> \}

*  配置Akka并生成ActorSystem，启动JobManager
 *  启动HA和metric相关服务
 *  在`startJobManagerActors()`方法中启动JobManagerActors，以及webserver，TaskManagerActor，ResourceManager等等
 *  阻塞等待终止
 *  集群通过LeaderService等选出JobManager的leader

3.2.3 JobManager启动Task

JobManager 是一个Actor，通过各种消息来完成核心逻辑：

> override def handleMessage: Receive = \{  
>   case GrantLeadership(newLeaderSessionID) =>  
>     log.info(s"JobManager $getAddress was granted leadership with leader session ID " +  
>       s"$newLeaderSessionID.")  
>     leaderSessionID = newLeaderSessionID  
>       
>     .......
> 
>

有几个比较重要的消息：

*  GrantLeadership 获得leader授权，将自身被分发到的 session id 写到 zookeeper，并恢复所有的 jobs
 *  RevokeLeadership 剥夺leader授权，打断清空所有的 job 信息，但是保留作业缓存，注销所有的 TaskManagers
 *  RegisterTaskManagers 注册 TaskManager，如果之前已经注册过，则只给对应的 Instance 发送消息，否则启动注册逻辑：在 InstanceManager 中注册该 Instance 的信息，并停止 Instance BlobLibraryCacheManager 的端口【供下载 lib 包用】，同时使用 watch 监听 task manager 的存活
 *  SubmitJob 提交 jobGraph   
    最后一项SubmintJob就是我们要关注的，从客户端收到JobGraph，转换为ExecutionGraph并执行的过程。

> private def submitJob(jobGraph: JobGraph, jobInfo: JobInfo, isRecovery: Boolean = false): Unit = \{  
>       
>     ......  
>       
>     executionGraph = ExecutionGraphBuilder.buildGraph(  
>           executionGraph,  
>           jobGraph,  
>           flinkConfiguration,  
>           futureExecutor,  
>           ioExecutor,  
>           scheduler,  
>           userCodeLoader,  
>           checkpointRecoveryFactory,  
>           Time.of(timeout.length, timeout.unit),  
>           restartStrategy,  
>           jobMetrics,  
>           numSlots,  
>           blobServer,  
>           log.logger)  
>             
>     ......  
>       
>     if (leaderElectionService.hasLeadership) \{  
>             log.info(s"Scheduling job $jobId ($jobName).")  
>               
>             executionGraph.scheduleForExecution()  
>               
>           \} else \{  
>             self ! decorateMessage(RemoveJob(jobId, removeJobFromStateBackend = false))
> 
>             log.warn(s"Submitted job $jobId, but not leader. The other leader needs to recover " +  
>               "this. I am not scheduling the job for execution.")  
>       
>     ......  
> \}

首先做一些准备工作，然后获取一个ExecutionGraph，判断是否是恢复的job，然后将job保存下来，并且通知客户端本地已经提交成功了，最后如果确认本JobManager是leader，则执行`executionGraph.scheduleForExecution()`方法，这个方法经过一系列调用，把每个ExecutionVertex传递给了Excution类的deploy方法：

> public void deploy() throws JobException \{
> 
>         ......
> 
>         try \{  
>             // good, we are allowed to deploy  
>             if (!slot.setExecutedVertex(this)) \{  
>                 throw new JobException("Could not assign the ExecutionVertex to the slot " + slot);  
>             \}
> 
>             // race double check, did we fail/cancel and do we need to release the slot?  
>             if (this.state != DEPLOYING) \{  
>                 slot.releaseSlot();  
>                 return;  
>             \}
> 
>             if (LOG.isInfoEnabled()) \{  
>                 LOG.info(String.format("Deploying %s (attempt \#%d) to %s", vertex.getTaskNameWithSubtaskIndex(),  
>                         attemptNumber, getAssignedResourceLocation().getHostname()));  
>             \}
> 
>             final TaskDeploymentDescriptor deployment = vertex.createDeploymentDescriptor(  
>                 attemptId,  
>                 slot,  
>                 taskState,  
>                 attemptNumber);
> 
>             final TaskManagerGateway taskManagerGateway = slot.getTaskManagerGateway();
> 
>             final CompletableFuture<Acknowledge> submitResultFuture = taskManagerGateway.submitTask(deployment, timeout);
> 
>             ......  
>         \}  
>         catch (Throwable t) \{  
>             markFailed(t);  
>             ExceptionUtils.rethrow(t);  
>         \}  
>     \}
> 
>

我们首先生成了一个TaskDeploymentDescriptor，然后交给了`taskManagerGateway.submitTask()`方法执行。接下来的部分，就属于TaskManager的范畴了。

直接看原文章吧

[https://www.cnblogs.com/bethunebtj/p/9168274.html\#133-%E7%A8%8B%E5%BA%8F%E5%90%AF%E5%8A%A8%E8%BF%87%E7%A8%8B][https_www.cnblogs.com_bethunebtj_p_9168274.html_133-_E7_A8_8B_E5_BA_8F_E5_90_AF_E5_8A_A8_E8_BF_87_E7_A8_8B]

[image_1caf1oll019fp1odv1bh9idosr79.png-486.3kB]: /images/20211203/e65b6773c2a94470b8cc767c9d2dd498.png
[2019072209225395.png]: /images/20211203/4ac14886a200442da0e941f655d57eff.png
[20190722092347166.png]: /images/20211203/52d84a846ad547c4812a312d20dce910.png
[20190722092522663.png]: /images/20211203/0dc2cbafc81f4af89c480154720866b0.png
[image_1caf64b7c1gjnv2eebi1v9e1cvum.png-129.4kB]: https://imgconvert.csdnimg.cn/aHR0cDovL3N0YXRpYy56eWJ1bHVvLmNvbS9iZXRodW5lYnRqLzY5djlzeXIycDVrNW9tM2M0am94OXdoMC9pbWFnZV8xY2FmNjRiN2MxZ2pudjJlZWJpMXY5ZTFjdnVtLnBuZw
[image_1caf6ak4rkqsc1u1hci93fe0d13.png-36.6kB]: /images/20211203/acc88808f5a14a40b63ca733156308bd.png
[image_1caf71h79s0s3fodem1aeb1j3m1g.png-83.8kB]: /images/20211203/1b2e483c327f4b6a8e2bc1e96fe21319.png
[20190722093136115.png]: /images/20211203/c430b9ec4b9448a99e3900ee6938a463.png
[Link 1]: http://flink.apache.org/visualizer/
[image_1cafgsliu1n2n1uj21p971b0h6m71t.png-25.7kB]: /images/20211203/ba94a4e718324213b9409c2bf4d02561.png
[image_1cafj7s6bittk5tt0bequlig2a.png-158.7kB]: /images/20211203/6efdd3476a514e59a38798fd5d0566a8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70]: /images/20211203/681b2f106e864ac4a92b348663a7c911.png
[image_1cafn516r1p68kt31g7r196rcsv2n.png-40.1kB]: /images/20211203/aa9ce6fca1c240e7bf18bf83eca15711.png
[image_1cafnu1pl1d8c15m219b8vkb2334.png-112.9kB]: /images/20211203/f099ecda599e498b8aa1e0da65f9ec79.png
[image_1cclle7ui2j41nf611gs1is18m19.png-127.5kB]: /images/20211203/aa80cfddd41d4b99af3278aa038605e3.png
[image_1cafpf21c1jh3s5ap1fisu4v23h.png-44.7kB]: /images/20211203/f8f78b640a7f4550acd87a13fac0f4c9.png
[image_1cafpko68b3r1lk0dpsnmbj3c3u.png-61.2kB]: /images/20211203/92ae46852ba04470ba685b855c00f83d.png
[image_1cafqroarkjkuje1hfi18gor654b.png-137kB]: /images/20211203/4c518452edc740f58dc457ac8e05753c.png
[https_www.cnblogs.com_bethunebtj_p_9168274.html_133-_E7_A8_8B_E5_BA_8F_E5_90_AF_E5_8A_A8_E8_BF_87_E7_A8_8B]: https://www.cnblogs.com/bethunebtj/p/9168274.html#133-%E7%A8%8B%E5%BA%8F%E5%90%AF%E5%8A%A8%E8%BF%87%E7%A8%8B