10-flink TaskManager 和 Slots

Posted 2023-04-17

参考技术A 思考问题:
1.怎么样实现并行计算？
答：设置并行度。多线程，不同任务放到不同线程上。

2.并行的任务，需要占用多少slot？

3.一个流处理程序，到底包含多少个任务？

一、TaskManager和Slot的关系介绍

process：进程
Treads：线程

二、并行度（parallelism）

每一个线程占用一个slot，上图一中任务合并为上图二所示（任务链，后续讲解），图中算子并行度最大的（算子后面的中括号数字代表并行度）为2，所以整个flink程序的并行度为2，所以只需要2个slot就可以跑起来。

 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中的宽依赖

图中：A4 代表 A任务有4个，C2表示C任务2个，以此类推

taskmanager.numberOfTaskSlots:3 每个taskmanager设置了并行度为3

设一共有3个TaskManager，每一个TaskManager中的分配3个TaskSlot，也就是每个TaskManager可以接收3个task，一共9个TaskSlot，如果我们设置parallelism.default=1，即运行程序默认的并行度为1，9个TaskSlot只用了1个，有8个空闲，因此，设置合适的并行度才能提高效率。

三、思考
假设当前可用的slot只有1个，任务有4个，slot不够用的时候，则会一直等待分配资源，直到超时报错。

slot推荐设置为当前机器的核心数，假设cpu核心数为4核，则设置4。
slot占用数量与并行度最大的算子一致。

flink 并行度

参考技术A Flink 作为一套分布式执行框架，计算资源可以不断的扩展。
不同的任务类型，可以控制需要的计算资源。在flink整个runtime的模型中
并行度是一个很重要的概念，通过设置并行度可以为认为分配合理的计算资源，
做到资源的合理配置。

整个flink的架构简单的说是 中心控制（jobManager）+ 多点分布执行（taskManager）
弹性的资源分配主要来自于taskManager的有效管理和配置。

在启动flink 之前，在核心的配置文件里面，需要指定两个参数。
taskmanager.numberOfTaskSlots 和 parallelism.default。

首先需要明白slot的概念。对于 taskManager，他其实是一个 JVM 程序。
这个JVM 可以同时执行多个task，每个task 需要使用本机的硬件资源。
slot 的属于 jvm 管理的 一些列资源卡槽。 每个slot 只能执行一个task。
每个slot分配有固定的内存资源，但是不做cpu的隔离。 JVM管理一个 slot的pool，
用来执行相应的task。taskmanager.numberOfTaskSlots = 10，则理论上可以同时执行10个子任务。

那么对于1个5节点，numberOfTaskSlots= 6的集群来说，那么就有30个slot可以使用。
对于具体的一个job来说，他会贪婪的使用所有的 slot吗？
使用多少slot 是由parallelism.default 决定的。如果是 5， 那么对于一个job 他最多同时使用5个slot。
这个配置对于多job平台的集群是很有必要的。

那么给定一个stream api 编写的flink 程序，被分解的task是否和map 到slot 上执行的呢？
flink 有几个经典的graph， stream-api对应的stream_graph-> job_graph->execution_graph->物理执行图。
execution_graph 基本就决定了如何分布执行。
我们知道一个 stream-api, 主要有 source, operate, sink 这几部分。那么我们可以从source开始看 并行的控制。

source 有并行source和 非并行。我们主要看并行，想类似与kafka 这种生成消费者模式的数据源，能够 并行消费source是非常重要的。
所以可以看到kafka，FlinkKafkaConsumerBase<T> extends RichParallelSourceFunction<T>，可以充分利用并行度，大大提高吞吐量。
对应到具体的物理执行上，就是多个 source task 任务执行，他们属于一个kafka group同时消费 不同的partition。
对于parallelSource，默认使用cpu 核心做并行度。我们可以通过api进行设置。

接下来是 operate，每个operate都可以设置parallel，如果没有设置将会使用其他层次的设置，比如env，flink.conf中的配置，parallelism.default。
比如 source. map1().map2().grouby(key).sink()
这样一个程序，默认，source和 map1，map2有同样的parallel，上游的output 可以直接one-one forwarding.
在flink 的 优化中，甚至可以把这些 one-one 的operate 合成一个，避免转发，线程切换，网络通信开销。
对于groupby 这样的算子，则属于另外的一类。上游的output 需要 partion 到下游的不同的节点，而不能做位一个chain。

由于operate可以设置独自的parallel，如果与上游不一致。上游的output必然需要某种partion策略来 rebalnce数据。kafka有很多策略来处理这个细节。
对于partion放在专门的章节来说明。
对于sink，则可以理解位一个特定的operate，目前看没什么特殊处理逻辑。