Flink运行架构_flink redistribute

4.1 运行架构
Flink Runtime 层的主要架构如下图所示，它展示了一个 Flink 集群的基本结构。整体来说，它采用了标准 master-slave主从的结构，master负责管理整个集群中的资源和作业；TaskManager 则是 Slave，负责提供具体的资源并实际执行作业

4.2 核心组件
Application Master 部分包含了三个组件:

1. Dispatcher
   负责接收用户提供的作业，并且负责为这个新提交的作业启动一个新的 JobManager 组件
2. ResourceManager
   负责资源的管理，在整个 Flink 集群中只有一个 ResourceManager
3. JobManager
   负责管理作业的执行，在一个 Flink 集群中可能有多个作业同时执行，每个作业 都有自己的 JobManager 组件
   还有其他组件：
4. TaskManager
   主要负责执行具体的task任务，从JobManager处接收需要部署的 Task，部署 启 动后，与自己的上游建立连接，接收数据并处理。
5. Cluster Manager
   集群管理器，比如Standalone、YARN、K8s等，就是前面我们学习的不同环境
6. Client
   提交Job的客户端，可以是运行在任何机器上（与 JobManager 环境连通即可）。提交Job后，Client可以结束进程（Streaming的任务），也可以不结束并等待结果返回
   4.3 核心概念
   4.3.1 TaskManager与Slots
   Flink中每一个worker(TaskManager)都是一个JVM进程，它可能会在独立的线程上执行一个Task。为了控制一个worker能接收多少个task，worker通过Task Slot来进行控制（一个worker至少有一个Task Slot）。
   这里的Slot如何来理解呢？很多的文章中经常会和Spark框架进行类比，将Slot类比为Core，其实简单这么类比是可以的，可实际上，可以考虑下，当Spark申请资源后，这个Core执行任务时有可能是空闲的，但是这个时候Spark并不能将这个空闲下来的Core共享给其他Job使用，所以这里的Core是Job内部共享使用的。接下来我们再回想一下，之前在Yarn Session-Cluster模式时，其实是可以并行执行多个Job的，那如果申请两个Slot，而执行Job时，只用到了一个，剩下的一个怎么办？那我们自认而然就会想到可以将这个Slot给并行的其他Job，对吗？所以Flink中的Slot和Spark中的Core还是有很大区别的。
   每个task slot表示TaskManager拥有资源的一个固定大小的子集。假如一个TaskManager有三个slot，那么它会将其管理的内存分成三份给各个slot。资源slot化意味着一个task将不需要跟来自其他job的task竞争被管理的内存，取而代之的是它将拥有一定数量的内存储备。需要注意的是，这里不会涉及到CPU的隔离，slot目前仅仅用来隔离task的受管理的内存。

4.3.2 Parallelism（并行度）
在学习Spark RDD时，无论是读取内存中的数据，或读取文件数据，都会接触一个叫并行度的概念，并且在RDD的算子中也可以动态改变并行度，通过学习，咱们应该知道Spark中的并行度最终体现为分区，而分区又意味着Task。所以Spark 计算中Task的数量是可以通过并行度推算出来的。这个大家没有的问题的话，那就好办了，为什么？因为Flink的并行度的作用和Spark中并行度的作用的一样的。最后都可以表现为任务的并行执行。幸福感满满的。
虽然Spark中的并行度和Flink的并行度的原理，作用基本一致，但是由于模型选择的问题，所以使用上依然有些细微的区别：
 Spark的并行度设置后需要调用特殊的算子（repartition）或特殊的操作（shuffle）才能进行改变，比如调用flatMap算子后再调用repartition改变分区。
 Flink的并行度设置可以在任何算子后使用，并且为了方便，也可以设置全局并行度
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(2);
但是需要注意，某些数据源数据的采集是无法改变并行度，如Socket
 如果Flink的一个算子的并行度为2，那么这个算子在执行时，这个算子对应的task就会拆分成2个subtask，发到不同的Slot中执行

一个特定算子的子任务（subtask）的个数被称之为其并行度（parallelism），一般情况下，一个流程序的并行度，可以认为就是其所有算子中最大的并行度。一个程序中，不同的算子可能具有不同的并行度。
Stream流在算子之间传输数据的形式可以是one-to-one(forwarding)的模式也可以是redistributing的模式，具体是哪一种形式，取决于算子的种类。
 One-to-one：stream(比如在source和map operator之间)维护着分区以及元素的顺序。那意味着flatmap 算子的子任务看到的元素的个数以及顺序跟source 算子的子任务生产的元素的个数、顺序相同，map、fliter、flatMap等算子都是one-to-one的对应关系。类似于spark中的窄依赖
 Redistributing：stream(map()跟keyBy/window之间或者keyBy/window跟sink之间)的分区会发生改变。每一个算子的子任务依据所选择的transformation发送数据到不同的目标任务。例如，keyBy()基于hashCode重分区、broadcast和rebalance会随机重新分区，这些算子都会引起redistribute过程，而redistribute过程就类似于Spark中的shuffle过程。类似于spark中的宽依赖
4.3.3 Task与SubTask
Flink执行时,由于并行度的设置，可以将同一个Job不同算子的subtask放在同一块内存中进行处理，那么这样在执行时就可以合并成一个完整的task进行处理，而不是独立的子任务，这样就减少了子任务（SubTask）之间调度和数据传递的性能损耗

4.3.4 Operator Chains（任务链）
在Flink执行计算时，多个算子的subTask到底能不能组成一个Task是不确定的。比如读取并行度为1的数据源，但是map映射时使用并行度2，那么这样map算子就存在两个subtask，可以数据源读取时只有一个subtask，那么就会导致其中一个subtask无法链接成task，就需要在其他slot中执行。所以在这种情况下，到底哪些subtask可以组合，哪些subtask不能组合，就需要动态调整，这就需要用到一种任务链的操作进行设置。

任务链必须满足两个条件：one-to-one的数据传输并且并行度相同
4.3.5 ExecutionGraph（执行图）
由Flink程序直接映射成的数据流图是StreamGraph，也被称为逻辑流图，因为它们表示的是计算逻辑的高级视图, 简单理解就是将整个流计算的执行过程用图形表示出来,这样更直观，更便于理解，所有用于表示程序的拓扑结构

虽然更便于理解，但是和真正执行还有差别的，因为到底什么样的subtask组合成一个完整的task，task之间如何将多个符合条件的节点 chain 在一起作为一个节点，这些还是不能直观的展示给我们，所以为了直观地观察一个流处理程序的执行，Flink还需要将逻辑流图转换为作业图 JobGraph，提交给 JobManager

JobManager 根据JobGraph生成ExecutionGraph。ExecutionGraph是JobGraph的并行化版本，是调度层最核心的数据结构
4.4 提交流程
我们来看看当一个应用提交执行时，Flink的各个组件是如何交互协作的：

图 任务提交和组件交互流程
上图是从一个较为高层级的视角，来看应用中各组件的交互协作。如果部署的集群环境不同（例如YARN，Mesos，Kubernetes，Standalone等），其中一些步骤可以被省略，或是有些组件会运行在同一个JVM进程中。
具体地，如果我们将Flink集群部署到YARN上，那么就会有如下的提交流程：

1. Flink任务提交后，Client向HDFS上传Flink的Jar包和配置
2. 向Yarn ResourceManager提交任务，ResourceManager分配Container资源
3. 通知对应的NodeManager启动ApplicationMaster，ApplicationMaster启动后加载Flink的Jar包和配置构建环境，然后启动JobManager
4. ApplicationMaster向ResourceManager申请资源启动TaskManager
5. ResourceManager分配Container资源后，由ApplicationMaster通知资源所在节点的NodeManager启动TaskManager
6. NodeManager加载Flink的Jar包和配置构建环境并启动TaskManager
7. TaskManager启动后向JobManager发送心跳包，并等待JobManager向其分配任务。