Flink架构及基本概念

TaskManager启动后&＃xff0c;JobManager会与它建立连接&＃xff0c;并将作业图&＃xff08;JobGraph&＃xff09;转换成可执行的执行图&＃xff08;ExecutionGraph&＃xff09;分发给可用的TaskManager&＃xff0c;然后由TaskManager具体执行任务。

作业管理器&＃xff08;JobManager&＃xff09;

对于一个提交执行的作业&＃xff0c;Jobmanager是管理者&＃xff08;Master&＃xff09;&＃xff0c;负责管理调度&＃xff0c;在不考虑高可用的情况下只能有一个。JobManager是一个Flink集群中任务管理和调度的核心&＃xff0c;是控制应用执行的主程序。即每个以用都应该被唯一的JobManager所控制执行。
JobManager包括3个组件

JobMaster

jobMaster是JobManager中最核心的组件&＃xff0c;负责处理单独的作业。jobMaster和具体的job是一一对应的&＃xff0c;多个job可以同时运行在一个Flink集群中&＃xff0c;每个job都有一个自己的jobMaster
作业提交时&＃xff0c;jobMaster会先接收到要执行的应用。jobMaster会把JobGraph转换成一个物理层面的数据流图&＃xff0c;这个图被称作“执行图”&＃xff0c;它包含了所有可以并发执行的任务。jobMaster会向资源管理器发出请求&＃xff0c;申请执行任务必要的资源。一旦获取到了足够的资源&＃xff0c;就会将执行图发到真正运行它们的TaskManager上。
在运行过程中&＃xff0c;JobMaster会负责所有需要中央协调的操作。

ResourceManager&＃xff08;资源管理器&＃xff09;

ResourceManager主要负责资源的分配和管理&＃xff0c;在Flink集群中只有一个。所谓“资源”&＃xff0c;主要指的是TaskManager的任务槽&＃xff08;task Slots&＃xff09;。任务槽就是Flink集群中的资源调配单元&＃xff0c;包含了机器用来执行计算的一组CPU和内存资源。每一个任务&＃xff08;task&＃xff09;都需要分配到一个slot上执行。
在Standalone部署时&＃xff0c;因为TaskManager是单独启动的&＃xff0c;所以ResourceManager只能分发可用TaskManager的任务槽&＃xff0c;不能启动新的TaskManager。
在有资源管理平台时&＃xff0c;当新的作业申请资源时&＃xff0c;ResourceManager会将有空闲的槽位的TaskManager分配给JobManager。如果ResourceManager没有足够的任务槽&＃xff0c;它还可以向资源提供平台发起会话&＃xff0c;请求提供启动TaskManager进程的容器。另外&＃xff0c;ResourceManager还负责停掉空闲的TaskManager&＃xff0c;释放计算资源。

Dispatcher&＃xff08;分发器&＃xff09;

Dispatcher主要负责提供一个REST接口&＃xff0c;用来提交作业&＃xff0c;并且负责每一个新提交的作业启动一个新的JobManager组件。Dispatcher也会启动一个Web UI&＃xff0c;用来方便地展示监控作业执行的信息。

任务管理器&＃xff08;TaskManager&＃xff09;

TaskManager是工作者&＃xff08;Worker、Slave&＃xff09;&＃xff0c;负责执行任务处理数据&＃xff0c;所以可以有一个或多个。
TaskManager是Flink中的工作进程&＃xff0c;负责数据流的具体计算任务。Flink集群中必须至少有一个TaskManaager&＃xff1b;每个TaskManager都包含了一定数量的任务槽。Slot是资源调度的最小单位&＃xff0c;slots的数量限制了TaskManager能够并行处理的任务数量。
启动之后&＃xff0c;TaskManager会将一个或多个槽位提供给JobMaster调用&＃xff0c;jobMaster就可以分配任务来执行。
在执行过程中&＃xff0c;TaskManager可以缓冲数据&＃xff0c;还可以跟其它运行同一应用的TaskManager交换数据。

作业提交流程

独立模式

独立模式下&＃xff0c;只有会话模式和应用模式两种部署方式。两者流程基本类似&＃xff1a;TaskManager都需要手动启动&＃xff0c;所以当ResourceManager收到JobMaster的请求时&＃xff0c;会直接要求TaskManager提供资源。而JobMaster的启动时间点&＃xff0c;会话模式是预先启动&＃xff0c;应用模式则是在作业提交时启动。
1. 由客户端通过分发器提供的REST接口&＃xff0c;将作业提交给JobManager
2. 由分发器启动 JobMaster&＃xff0c;并将作业&＃xff08;包含 JobGraph&＃xff09;提交给 JobMaster。
3. JobManager将JobGraph解析为可执行的ExecutionGraph&＃xff0c;得到所需的资源数量&＃xff0c;然后再向资源管理器请求任务资源槽。
4. 资源管理器判断当前是否有足够的资源&＃xff1b;如果没有则启动新的TaskManager。
5. TaskManager启动后&＃xff0c;向ResourceManager注册自己可用的任务槽。
6. 资源管理器通知TaskManager为新的作业提供slots。
7. TaskManager连接到对应的JobMaster&＃xff0c;提供slots
8. jobMaster将需要执行的任务分发给TaskManager。
9. TaskManager执行任务。

Yarn集群

• 会话模式 提交作业流程
  1. 客户端通过REST接口&＃xff0c;将作业提交给分发器
  2. 分发器启动JobMaster&＃xff0c;并将作业&＃xff08;包含JobGraph&＃xff09;提交给JobMaster
  3. JobMaster向资源管理器请求资源
  4. 资源管理器向Yarn的资源管理器请求container资源
  5. Yarn启动新的TaskManager容器
  6. TaksManager启动后&＃xff0c;向Flink的资源管理器注册自己的可用任务槽
  7. 资源管理器通知TaskManager为新的作业提供任务槽
  8. TaskManager连接到对应的JobMaster&＃xff0c;提供任务槽
  9. JobMaster将需要执行的任务分发到TaskManager&＃xff0c;执行任务

• 单作业模式提交作业流程
  1. 客户端将作业提交给YARN的资源管理器&＃xff0c;这一步中会同时将Flink的Jar包和配置上传到HDFS&＃xff0c;以便后续启动的Flink相关组件的容器
  2. YARN的资源管理器分配容器资源&＃xff0c;启动Flink JobManager&＃xff0c;并将作业提交给JobMaster
  3. JobMaster向资源管理器请求资源
  4. 资源管理器请YARN的资源管理器请求容器
  5. YARN启动新的TakManager容器
  6. TaskManager启动后&＃xff0c;向Flink的任务管理器注册自己的可用任务槽
  7. 资源管理器通知TaskManager为新的任务提供任务槽
  8. TaskManager连接到对应的JobMaster&＃xff0c;提供任务槽
  9. JobManager将需要执行的任务分发给TaskManager&＃xff0c;执行任务

• 应用模式
  应用模式与单作业模式的提交流程类似&＃xff0c;只是初始提交给YARN资源管理器的不再是具体的作业&＃xff0c;而是整个应用。一个应用中可能包含多个作业&＃xff0c;这些作业都在Flink集群中启动各自对应的JobMaster。

重要概念

数据流图&＃xff08;Dataflow Graph&＃xff09;

Flink中每一个处理转换操作称为“算子”。
所有的Flink程序都可以归纳为由三部分构成&＃xff1a;Source、Transformation和Sink

• Source 表示“元算子”&＃xff0c;负责读取数据源。
• Transformation表示“转换算子”&＃xff0c;利用各种算子进行加工处理
• Sink表示“下沉算子”&＃xff0c;负责数据的输出

在运行时&＃xff0c;Flink程序会被映射成所有算子按照逻辑顺序链接在一起的一张图&＃xff0c;被称为“逻辑数据流”&＃xff08;数据流图&＃xff09;。

并行度

一个特定算子的子任务&＃xff08;subtask&＃xff09;的个数被称之为其并行度&＃xff08;parallelism&＃xff09;。包含并行子任务的数据流&＃xff0c;就是并行数据流&＃xff0c;它需要多个分区&＃xff08;stream partition&＃xff09;来分配并行任务。一般情况下&＃xff0c;一个流程序的并行度&＃xff0c;可以认为就是其所有算子中最大的并行度。一个程序中&＃xff0c;不同的算子可能具有不同的并行度

并行度设置

Flink中&＃xff0c;可以用不同的方法来设置并行度&＃xff0c;它们的有效范围和优先级是不同的。

1. 代码中设置

// 算子后跟着调用setParallelism()方法&＃xff0c;来设置当前算子的并行度&＃xff0c;这种设置只对当前算子有效
stream.map((_, 1)).setParallelism(2)


注&＃xff1a;KeyBy()方法返回的不是算子&＃xff0c;所以无法对KeyBy()设置并行度
2. 提交作业时设置

# 使用flink run命令提交作业时&＃xff0c;可以增加-p参数来设置当前应用程序的执行并行度&＃xff0c;作用类似于执行环境的全局设置
bin/flink run -p 2 -c com.wc.StreamWordCount ./FlinkWc-1.0.jar


3. 配置文件中设置
   修改集群的配置文件flink-conf.yaml中直接更改默认并行度

parallelism.default: 2


对整个集群上提交的所有作业有效&＃xff0c;默认为1。

算子连

在Flink中&＃xff0c;并行度相同的一对一算子操作&＃xff0c;可以直接连接在一起 形成一个“大”的任务&＃xff0c;这样原本的算子就成为真正任务里的一部分&＃xff0c;每个任务会被一个线程执行。这样的技术被称为“算子链”。

算子间数据传输

1. 一对一
   这种模式下&＃xff0c;数据流维护着分区以及元素的顺序。Source算子读取数据之后&＃xff0c;可以直接发送给 map()算子做处理&＃xff0c;它们之间不需要重新分区&＃xff0c;也不需要调整数据的顺序。这就意味着 map() 算子的子任务&＃xff0c;看到的元素个数和顺序跟 Source 算子的子任务产生的完全一样&＃xff0c;保证着“一对一”的关系。map()、filter()、flatMap()等算子都是这种one-to-one 的对应关系
2. 重分区
   每一个算子的子任务&＃xff0c;会根据数据传输的策略&＃xff0c;把数据发送到不同的下游目标任务。例如&＃xff0c;keyBy()是分组操作&＃xff0c;本质上基于键&＃xff08;key&＃xff09;的哈希值&＃xff08;hashCode&＃xff09;进行了重分区&＃xff1b;而当并行度改变时&＃xff0c;比如从并行度为 2 的 window 算子&＃xff0c;要传递到并行度为 1 的 Sink 算子&＃xff0c;这时的数据传输方式是再平衡&＃xff08;rebalance&＃xff09;&＃xff0c;会把数据均匀地向下游子任务分发出去。这些传输方式都会引起重分区&＃xff08;redistribute&＃xff09;的过程

合并算子链

Flink 默认会按照算子链的原则进行链接合并&＃xff0c;如果我们想要禁止合并或者自行定义&＃xff0c;也可以在代码中对算子做一些特定的设置&＃xff1a;

// 禁用算子链
.map((_,1)).disableChaining()
// 从当前算子开始新链
.map((_,1)).startNewChain()


作业图与执行图

Flink 中任务调度执行的图&＃xff0c;按照生成顺序可以分成四层&＃xff1a;
逻辑流图&＃xff08;StreamGraph&＃xff09;→ 作业图&＃xff08;JobGraph&＃xff09;→ 执行图&＃xff08;ExecutionGraph&＃xff09;→ 物理图&＃xff08;Physical Graph&＃xff09;

1. 逻辑流图
   根据用户通过DataStream API编写的代码生的最初的DAG图&＃xff0c;用来表示程序的拓扑结构
2. 作业图
   逻辑流图经过优化之后生成的就是作业图&＃xff0c;这是提交给JobManager的数据结构&＃xff0c;确定了当前作业中所有任务的划分。主要的优化为&＃xff1a;将多个符合条件的节点连接在一起合并成一个任务节点&＃xff0c;形成算子链&＃xff0c;这样可以减少数据交换的消耗。作业图一般在客户端生成&＃xff0c;在作业提交时传递给JobMaster。
3. 执行图
   JobMaster收到作业图后&＃xff0c;会根据它来生成执行图。执行图时作业图的并行化版本&＃xff0c;是调度最核心的数据结构。
4. 物理图
   JobMaster生成执行图后&＃xff0c;会将它分发给TaskManager&＃xff1b;各个TaskManager会根据执行图部署任务&＃xff0c;最终的物理执行过程也会形成一张“图”&＃xff0c;一般就叫做物理图&＃xff0c;这只是具体执行层面的图&＃xff0c;并不是一个具体的数据结构

任务和任务槽

任务槽

Flink中每一个work都是一个JVM进程&＃xff0c;它可以启动多个独立的线程&＃xff0c;来并行执行多个任务。
为了控制并发量&＃xff0c;需要在TaskManager上对每个任务运行所占用的资源做出明确的划分&＃xff0c;这就是所谓的任务槽。每个任务槽其实表示了TaskManager拥有计算资源的一个固定大小的子集。

任务槽数量的设置

可以通过集群的配置文件来设定TaskManager的槽数量

taskmanager.numberOfTaskSlots: 8


通过调整槽的数量&＃xff0c;可以控制子任务间的隔离级别。槽目前只能用来隔离内存&＃xff0c;不会涉及CPU隔离

任务对任务槽的共享

默认情况下&＃xff0c;Flink允许子任务共享槽。如果希望某个算子对应的任务完全独占一个槽&＃xff0c;或者只有某一部分共享槽&＃xff0c;可以通过设置“槽共享组”手动指定

.map((_,1)).slotSharingGroup("1")


这样属于同一个槽共享组的子任务&＃xff0c;才会开启槽共享&＃xff1b;不同组之间的任务时完全隔离的&＃xff0c;必须分配到不同的槽上

任务槽和并行度的关系

任务槽时静态的概念&＃xff0c;是指TaskManager具有的并发执行能力&＃xff0c;可以通过参数taskmanager…numberOfTaskSlots进行设置&＃xff1b;而并行度是动态概念&＃xff0c;就是TaskManager运行程序时实际使用的并发能力&＃xff0c;可以通过参数parallelism.default进行设置。