【Flink---04】Flink运行时架构 『 系统架构 | 作业提交流程 | 数据流图 vs 作业图 vs 执行图 vs 物理图 | 并行度 | 任务槽| 算子任务 | 算子子任务』_flink 的作业提交和任务处理系统

1. 系统架构

1.1 概述

Flink运行时架构中，最重要的就是两大组件：

1. JobManager：管理者
2. TaskManager：工作者

Flink 的作业提交和任务处理时的系统图如下：

注意：

• 一般情况下(会话模式或单节点模式下)，客户端并不是处理系统的一部分，它只负责作业的提交。具体来说，就是调用程序的 main 方法，将代码转换成“数据流图”（Stream Graph），然后经过优化得到 作业图（JobGraph），并发送给 JobManager。提交之后，任务的执行其实就跟客户端没有关系了；我们可以在客户端选择断开与 JobManager 的连接, 也可以继续保持连接。之前我们在命令提交作业时，加上的-d 参数，就是表示分离模式（detached mode)，也就是断开连接。
• TaskManager 启动之后，JobManager 会与它建立连接，并将作业图（JobGraph）转换成“执行图”（ExecutionGraph）分发给可用的 TaskManager，然后就由 TaskManager 具体执行任务。

1.2 JobManager

一般情况下，一个Flink应用程序都应该被唯一的 JobManager 所控制执行。当然，在高可用（HA）的场景下，一个Flink应用程序含有多个job，因此会出现一个应用程序对应多个 JobManager的情况，此时只有一个是正在运行的领导节点（leader），其他都是备用节点（standby）。

1. 其主要职责有：
   1. 决定何时调度下一个Task或一组Task
   2. 对完成的Task或失败的Task做出反应
   3. 协调checkpoint
   4. 协调故障恢复
2. JobManger 又包含 3 个不同的组件：
   1. Dispatcher(分发器)：为每一个新提交的job启动一个JobMaster【如果设置了WebUI端口，还会启动 Web UI，用来方便地展示和监控作业执行的信息】
   2. JobMaster：负责一个Job的执行 / 任务调度。【一个JobMaster处理一个job，他们是一对一的关系】【JobMaster 会负责所有需要中央协调的操作，比如说检查点（checkpoints）
      的协调】
   3. ResourceManager：负责资源的分配和管理。主要是负责 任务槽（task slots） 的分配。

1.3 TaskManager

TaskManager：负责数据计算，真正处理作业的地方。

• Flink 中每一个 TaskManager都是一个 JVM 进程，因此每一个 TaskManager可以启动多个独立的线程，而每一个线程就对应一个子任务。

  注意：每一个算子对应一个任务(注意多个算子合并算一个算子，对应一个任务)，而每一个任务 (task) 拆分成多个并行的子任务（subtask）执行。

• 每一个 TaskManager 都包含一定数量的任务槽（task slots）。

注意：TaskManager的内存等资源大小可通过flink-conf.yaml文件中设置。

2. 作业提交流程

2.1 抽象流程

步骤如下：

1. 客户端将jar包程序 --> DataFlow Graph --> JobGraph，然后将JobGraph通过分发器提供的 rest 接口将作业提交给JobManager。
2. 然后分发器启动 JobMaster，并将JobGraph提交给 JobMaster。
3. JobMaster 将 JobGraph 解析为可执行的 ExecutionGraph，计算所需的资源数量，然后向ResourceManager请求资源（slots）。
4. ResourceManager判断当前是否由足够的可用资源，转到第6步；如果没有，启动新的 TaskManager。
5. TaskManager 启动之后，向 ResourceManager 注册自己的可用任务槽（slots）。
6. ResourceManager通知 TaskManager 为新的作业提供 slots
7. TaskManager 连接到对应的 JobMaster，提供 slots。
8. JobMaster 将需要执行的任务分发给 TaskManager。
9. TaskManager 执行任务，在执行过程中，TaskManager 可以缓冲数据，还可以跟其他运行同一应用的 TaskManager交换数据。

注意：如果集群环境不同（例如 Standalone、YARN、K8S 等），其中一些步骤可能会不同。所以接下来分别介绍不同部署下的流程。

2.2 Yarn集群下流程(动态分配TaskManager⭐)

2.2.1 会话模式

1. 需要预先启动一个 YARN session，来启动JobManager。内部流程如下：
   

   注意：

   1. 由于还没有提交作业，所以只有 JobManager中JobMaster没有运行
   2. TaskManager没有启动，后面是根据需要动态地启动

2. 发生job请求时的处理流程：
   
   1. 客户端将jar包程序 --> DataFlow Graph --> JobGraph，然后将JobGraph通过分发器提供的 rest 接口将作业提交给JobManager。
   2. 由分发器启动 JobMaster，并将 JobGraph 提交给 JobMaster。
   3. JobMaster 将 JobGraph 解析为可执行的 ExecutionGraph，计算所需的资源数量，然后向ResourceManager请求资源（slots）。
   4. ResourceManager判断当前是否由足够的可用资源，转到第6步；
      如果没有，启动新的 TaskManager。
      如果没有，ResourceManager 向 YARN ResourceManager 请求 container 资源。

      • 因为yarn集群是动态分配TaskManager，只要还有TaskManager能够启动且资源足够就能执行job
      • 因为yarn集群是动态回收TaskManager，因此，不会出现任务槽空闲的情况

   5. YARN 分配container 资源并启动新的 TaskManager 容器，一个container对应一个TaskManager
   6. TaskManager 启动之后，向 ResourceManager 注册自己的可用任务槽（slots）。
   7. ResourceManager通知 TaskManager 为新的作业提供 slots
   8. TaskManager 连接到对应的 JobMaster，提供 slots。
   9. JobMaster 将需要执行的任务分发给 TaskManager。
   10. TaskManager 执行任务，在执行过程中，TaskManager 可以缓冲数据，还可以跟其他运行同一应用的 TaskManager交换数据。

2.2.2 单作业模式

1. 不需要预先启动jobManager，job提交时会自动启动
2. 发生job请求时的处理流程：和上面的会话模式一样。

注意：可见，单作业模式与会话模式只是JobManager 的启动时间点不一样，会话模式是预先启动，应用模式则是在作业提交时启动。

2.2.3 应用模式

应用模式与单作业模式的提交流程一样，只是初始提交给 YARN ResourceManager的不再是具体的作业，而是整个应用。所以 程序–>DataFlowGraph–>JobGraph发生在jobMaster中而不是客户端中。

注意: 一个应用中可能包含了多个job，这些job都将在 Flink 集群中启动各自对应的 JobMaster。

2.3 Standalone集群下流程(预先固定TaskManager个数⭐)

在Standalone集群下，只有会话模式和应用模式两种部署方式。

• 两者的流程一样：
  1. 客户端将jar包程序 --> DataFlow Graph --> JobGraph，然后将JobGraph通过分发器提供的 rest 接口将作业提交给JobManager。
  2. 然后分发器启动 JobMaster，并将JobGraph提交给 JobMaster。
  3. JobMaster 将 JobGraph 解析为可执行的 ExecutionGraph，计算所需的资源数量，然后向ResourceManager请求资源（slots）。
  4. ResourceManager判断当前是否由足够的可用资源，转到第6步；
     如果没有，启动新的 TaskManager。
     如果没有，则不能执行该job，发出资源不足的异常

     • 因为预先固定TaskManager个数，不足就会失败
     • 因为预先固定TaskManager个数，不可动态分配也不可回收，所以可能出现任务槽空闲的情况

  5. TaskManager 启动之后，向 ResourceManager 注册自己的可用任务槽（slots）。
  6. ResourceManager通知 TaskManager 为新的作业提供 slots
  7. TaskManager 连接到对应的 JobMaster，提供 slots。
  8. JobMaster 将需要执行的任务分发给 TaskManager。
  9. TaskManager 执行任务，在执行过程中，TaskManager 可以缓冲数据，还可以跟其他运行同一应用的 TaskManager交换数据。
• 两者的区别：还是JobManager 的启动时间点不一样，会话模式是预先启动，应用模式则是在作业提交时启动。

3. 一些重要概念

3.1 算子

Flink 程序由算子组成，而算子分为3类：

• Source ：“源算子”，负责读取数据源。
• Transformation ： “转换算子”，负责数据计算。
• Sink ：“下沉算子”，负责数据的输出。

注意：

1. 每一条数据流（dataflow）以一个或多个source算子开始，经过一个或多个transformation算子转换，最后以一个或多个sink算子结束
2. 可以根据返回值类型来判断是否是算子：若返回值类型为SingleOutputStreamOperator或其子类。
3. 问：dataflow 中的算子和程序中基于 DataStream API 的方法是一一对应的关系吗？
   答：大多数情况下一一对应的关系，但是也存在多个方法对应一个算子的情况，比如wordCount程序中的.keyBy()返回的数据类型是 KeyedStream，而算子返回的数据类型是 SingleOutputStreamOperator，也就是说.keyBy().sum()一起组成了一个算子。

3.2 并行度（Parallelism）

(1) 什么是并行度？

• 算子的并行度：把一个算子操作，“复制”多份到多个结点，数据来了之后就可以到其中任意一个执行。这样一来，一个算子任务就被拆分成了多个并行的“子任务”(即多个线程执行)。而 算子的子任务个数就是该算子的并行度
• 一个流程序的并行度，就是其所有算子中最大的并行度。比如下面的流程序并行度为2：
  

(2) 怎么设置算子的并行度？

1. 在IDEA中设置并行度

   1. 为每个算子单独设置并行度：算子.setParallelism(并行度)
      

   2. 设置全局并行度(每个算子都是这个)：执行环境..setParallelism(并行度)
      

2. 提交应用时设置：只能设置全局并行度。【如果要设置全局并行度，一般用此方法，因为在代码中数字就写死了】

   1. 直接上传到集群

      语法：flink run –p 并行度 –c 入口类全类名 jar包路径
      
      eg：bin/flink run –p 2 –c com.atguigu.wc.StreamWordCount ./FlinkTutorial-1.0-SNAPSHOT.jar
      1
      2
      3

   2. 通过WebUI上传到集群
      

3. 配置文件中设置：只能设置全局并行度。在flink/conf/flink-conf.yaml文件中直接更改默认并行度

   parallelism.default: 2
   1

注意：这三种设置并行度的方式，如果组合设置，则从1, 2, 3优先级逐步降低。

(3) 关于并行度的优化：合并算子链

算子在关系上分为两类：

1. 一对一（One-to-one）算子：有map、filter、flatMap 等算子。这种算子如果相连，则最好将他们的并行度设置为相同，这样Flink在底层会做合并算子链的优化。

   • 合并算子链：在 Flink 中，前后链接的并行度相同的一对一（one to one）算子操作，对应的子任务直接合并为一个“大”的子任务（subtask）
     
   • 为什么合并？可以减少线程之间的切换和基于缓存区的数据交换，在减少时延的同时提升吞吐量
   • Flink底层自动合并算子链，如果禁止合并算子链？
     1. 方式一：前面的算子.disableChaining()，比如env.readTextFile().disableChaining()
     2. 方式二：后面的算子.startNewChain(),比如env.readTextFile().map()..startNewChain()

2. 重分区（Redistributing）算子：几个方法组成的算子会重写分区。比如.keyBy().sum()算子，keyBy()是分组操作会基于标签重新分区。

   • 重分区会怎样？涉及到subtask的交互，前一个算子的task会与当前重分区算子的所有tsak进行交互，而一对一算子不会。
     

3.3 数据流图 vs 作业图 vs 执行图 vs 物理图

1. 数据流图（StreamGraph）：在运行时，Flink 程序按照 算子逻辑 生成一个有向无环图DAG），这个图就是数据流图（StreamGraph）

   • 提示：提交作业之后，打开 Flink 自带的 Web UI，点击作业就能看到对应的 dataflow。比如：
     

2. 作业图（JobGraph）：StreamGraph 经过优化后生成的就是JobGraph。主要的优化为: 合并算子链

3. 执行图（ExecutionGraph）：在JobGraph的基础上结合各算子的并行度生成的就是ExecutionGraph 【ExecutionGraph与 JobGraph 最大的区别就是按照并行度将每个算子的任务拆分成了多个并行的子任务，并明确了任务间数据传输的方式。】

4. 物理图（Physical Graph）：TaskManager收到ExecutionGraph后将其转化为可执行的物理图（二进制代码看不懂）

例子：

3.4 任务（Tasks）和任务槽（Task Slots）

3.4.1 任务（Tasks）

一个算子任务被拆分成了多个并行的“子任务”(即多个线程执行)。而子任务个数就是该算子的并行度。

2.4.2 任务槽（Task Slots）

1. 什么是任务槽？

   • TaskManager 的计算资源是有限的，并行的任务越多，每个线程的资源就会越少。那么将TaskManager资源等额划分为几份？会根据程序设置的 任务槽数值 进行等额划分，等份后的每一份资源就是一个任务槽。
   • 【比如一个 TaskManager 有三个 slot，那么它会将管理的内存、cpu等资源平均分成三份，每个 slot 独自占据一份。】

     注意：建议将任务槽数目设置为主机的CPU核数，因为这样能避免对CPU资源的竞争

2. 如何设置任务槽数值？

   • 集群的配置文件中设置：

     taskmanager.numberOfTaskSlots: 8
     1

3.4.3 一个任务槽对应一个子任务吗？

• 不是的。默认情况下，Flink 是允许子任务共享 slot 的。具体是将不同任务的子任务放在同一个卡槽中。因此，一个 slot 上能够执行多个子任务。这样做有两个好处：

  1. 使得任务槽间就负载均衡，提高计算效率。

     比如：

     • 现有一执行图如下：
       
     • 假设是相同任务的子任务放在同一个任务槽中，则会造成任务槽间负载不均衡：
       
     • 假设是不同任务的子任务放在同一个任务槽中，这样任务槽间就负载均衡：
       

  2. 每一个任务槽都是完整的作业管道，这样一来，即使某个 TaskManage出现故障宕机，其他节点也可以完全不受影响，作业的任务可以继续执行。

     比如：
     

• 如果要设置某组算子任务间可以共享，而不是所有算子任务共享。则可以设置共享组：

  .map(word -> Tuple2.of(word, 1L)).slotSharingGroup(“1”); // 共享组数值相同的算子任务间可以共享
  1

3.4.4 判断job能否在任务槽上执行

如果job的并行度，即max{所有算子的并行度}，<= 所有TaskManager的可用任务槽总数，则该job可以执行；否则，不可以执行。