巨人肩膀
改版通知

巨人肩膀网站已全新改版。若您仍依赖旧站功能或数据，欢迎联系我们，我们会协助处理。联系我们

← 返回专栏列表

flink的常见知识点总结一

ckckck2025年1月10日7 浏览

1. Slot、TaskManager、并行度

1.1 TaskManager/Slots与CPU的关系

  • Flink的每个TaskManager为集群提供slot。插槽的数量通常与每个TaskManager的可用CPU内核数成比例。一般情况下，slot数等于每个TaskManager的CPU核数。
  • 经验上讲，Slot的数量与CPU-core的数量一致为好。但考虑到超线程，可以让slotNumber = 2 * cpuCore。

1.2 Slot与并行度

  • 一般我们设置task的并行度不能超过slot的数量。
  • 一个Task的并行度等于分配给它的Slot个数（前提槽资源充足）。

1.3 TaskManager、Slot、并行度之间的关系

  • 在Yarn集群中Job分离模式下，TaskManager的数量 = ceil(slot数量 / 并行度)。slotNumber >= TaskManager * 并行度。

1.4 一些概念

  1. Task：一个task可以看成是一条完整的算子链（source -> sink），在默认情况下，且线程够用的情况下，一个task运行在一个slot里面。
  2. Subtask：task的组成单位，一个subtask由一个线程来运行。
  3. Operator Chain：subtask中的算子连成链。
  4. 形成Subtask的规则：
    • 当数据发生重定向时，例如KeyBy操作。
    • 当程序算子并行度发生变化时。
    • 通过算子手动切分Operator（disableOperatorChaining, startNewChain, disableChaining）。
  5. Slot：隔绝内存，不隔绝CPU，意味着一个slot可以有多个线程。所以，一个slot里面可以运行多个subtask，这是一种优化，相同资源组在同一个slot。
  6. disableOperatorChaining：默认为开启，通过env.disableOperatorChaining()禁用。
  7. startNewChain：从当前算子之前开始划分一条新的Operator chain，在算子之后调用。
  8. disableChaining：针对某一个算子操作，断开算子前后的Operator chain，使算子单独作为一个subtask。
  9. 共享资源槽（slotSharingGroup("name")）：相同资源组的subtask在一个slot执行。如果算子调用之后，从这个算子开始，接下来的算子在一个组内（slot内）。

1.5 Task和Subtask

  • 一个task就代表一个operator，同样的一个算子链也是一个task。
  • 当运行Flink job时，一个task根据本身的并行度创建多个subtask，即subtask的数量代表这个task的并行度。
  • 每个subtask需要一个线程来运行。
  • 简单地，计算一个job的并行度：所有Operator和它的并行度的乘积。

1.6 Chain的含义与作用

  • Flink会尽可能地将多个operator链接（chain）在一起形成一个task pipeline。每个task pipeline在一个线程中执行。
  • 优点：它能减少线程之间的切换，减少消息的序列化/反序列化，减少数据在缓冲区的交换（即降低本地数据交换成本），减少了延迟的同时提高整体的吞吐量。提升了任务的执行效率。

1.7 形成算子链的条件

  • 同一个slot共享组；
  • 并行度相同；
  • one-to-one操作（map、filter、flatMap等算子都是one-to-one的对应关系）。
  • 上下游的并行度一致（槽一致）；
  • 该节点必须要有上游节点跟下游节点；
  • 下游StreamNode的输入StreamEdge只能有一个；
  • 上下游节点都在同一个slot group中；
  • 下游节点的chain策略为ALWAYS（可以与上下游链接，map、flatmap、filter等默认是ALWAYS）；
  • 上游节点的chain策略为ALWAYS或HEAD（只能与下游链接，不能与上游链接，Source默认是HEAD）；
  • 上下游算子之间没有数据shuffle（数据分区方式是forward）；
  • 用户没有禁用chain。

1.8 One-to-One

  • Stream维护着分区以及元素的顺序（比如source和map之间）。这意味着map算子的子任务看到的元素的个数以及顺序跟source算子的子任务生产的元素的个数、顺序相同。map、filter、flatMap等算子都是one-to-one的对应关系。

1.9 Redistributing

  • Stream的分区会发生改变。每一个算子的子任务依据所选择的transformation发送数据到不同的目标任务。例如，keyBy基于hashCode重分区、而broadcast和rebalance会随机重新分区，这些算子都会引起redistribute过程，而redistribute过程就类似于Spark中的shuffle过程。
  • 并行度不同的算子之前传递数据会进行重分区，Redistributing类型的算子也会进行重分区。

1.10 Slot

  • Slot其实是一个线程，一个subtask使用一个slot的资源去运行，一个task由多个subtask组成；
  • 因为有slot共享，Flink可以将资源占用少和资源占用多的subtask放到一起，这样能充分利用资源；
  • 有了slot共享，可以认为并行度最大（maxNum个subtask）的task所对应的并行度就是这个job所需的并行度。

1.11 TaskManager与Slot

  • TaskManager是一个JVM进程，并会以独立的线程来执行一个task。为了控制一个TaskManager能接受多少个task，Flink提出了Task Slot的概念。
  • 通过调整task slot的数量，用户可以定义task之间是如何相互隔离的。每个TaskManager有一个slot，也就意味着每个task运行在独立的JVM中。每个TaskManager有多个slot的话，也就是说多个task运行在同一个JVM中。而在同一个JVM进程中的task，可以共享TCP连接（基于多路复用）和心跳消息，可以减少数据的网络传输。也能共享一些数据结构，一定程度上减少了每个task的消耗。

1.12 Subtask与TaskSlot

  • Subtask是调度的基本单元；
  • TaskSlot是静态的概念，代表着TaskManager具有的并发执行能力；
  • Parallelism是动态的概念，是指程序运行时实际使用的并发能力；
  • Flink中slot是任务执行所申请资源的最小单元，同一个TaskManager上的所有slot都只是做了内存分离，没有做CPU隔离。

1.13 Slot槽位共享

1.13.1 简单的说

  • 为了高效使用资源，Flink默认允许一个job中不同task的subtask运行到一个slot中，就是Slot共享。

1.13.2 共享的意义

  • 因为每个subtask消耗的资源不同，将资源占用少和资源占用多的subtask放到一起，而不是都高或者都低的subtask放到一起，这样能使得资源能得到充分利用。比如source和sink一般都是网络通讯的IO操作，相对于中间Operator，消耗资源并不大。
  • 默认slot sharing是开启的，开启slot有两个好处：
    1. 计算一个job所需要的并行度：并行度最大的task所对应的并行度就是这个job所需的并行度。
    2. 更好的资源利用：将资源占用不多的source/map和资源占用较多的window算子放到一个slot，并且实现了占用资源较多的算子均匀分到了每个slot中，最后还有一个slot下跑了一整个pipeline。

1.13.3 同一个Job下的不同Task可以放到同一个Slot中——处理槽共享分组

1.14 Slot槽位共享机制

  • Flink在调度任务分配Slot的时候遵循两个重要原则：
    1. 同一个Job中的同一分组中的不同Task可以共享同一个Slot；
    2. Flink是按照拓扑顺序依次从Source调度到sink。

1.14.1 CoLocationGroup

  • 强制将subTasks放到同一个slot中，是一种硬约束：
    • 保证把JobVertices的第n个运行实例和其他相同组内的JobVertices第n个实例运作在相同的slot中（所有的并行度相同的subTasks运行在同一个slot）；
    • 主要用于迭代流（训练机器学习模型），用来保证迭代头与迭代尾的第i个subtask能被调度到同一个TaskManager上。

1.14.2 SlotSharingGroup

  • 它是Flink中用来实现slot共享的类，尽可能的允许不同的JobVertices部署在相同的Slot中，但这是一种宽约束，只是尽量做到不能完全保证。
    • 算子的默认group为default，所有任务可以共享同一个slot；
    • 要想确定一个未做SlotSharingGroup设置的算子的group是什么，可以根据上游算子的group和自身是否设置group共同确定（也就是说如果下游算子没有设置分组，它继承上游算子的分组）；
    • 为了防止不合理的共享，用户可以通过提供的API强制指定operator的共享组。因为不合理的共享槽资源（比如默认情况下所有任务共享所有的slot）会导致每个槽中运行的线程数增多，增加了机器负载。所以适当设置可以减少每个slot运行的线程数，从而整体上减少机器的负载。比如：someStream.filter(...).slotSharingGroup("group1")就强制指定了filter的slot共享组为group1。

1.14.3 任务链与处理槽共享组

  • 前者是对执行效率的优化，后者是对内存资源的优化。
  • 经验：调小TaskManager的资源（CPU和Memory），作业可以更好地分布。

1.15 Flink中常见异常解决

1.15.1 has no more allocated slots参数

  • akka.ask.timeout = 600s

1.15.2 java.io.IOException: Too many open files

  • state.backend.rocksdb.files.open = -1

1.15.3 泛型擦除错误

  • org.apache.flink.api.common.function.InvalidTypesException: The generic type parameters of '<class>' are missing
    • .returns(TypeInformation.of(new TypeHint<Tuple2<String, String>>() { }));
    • .returns(TypeInformation.of(WordWithCount.class));

1.15.4 akka.pattern.AskTimeoutException: Ask timed out on [Actor[akka://...]] after [10000 ms]

  • Akka超时导致，一般有两种原因：
    1. 集群负载比较大或者网络比较拥塞；
    2. 业务逻辑同步调用耗时的外部服务。
  • 如果负载或网络问题无法彻底缓解，需考虑调大akka.ask.timeout参数的值（默认只有10秒）；另外，调用外部服务时尽量异步操作（Async I/O）。

1.15.5 Checkpoint <cp_id> expired before completing

  • 首先应检查CheckpointConfig.setCheckpointTimeout()方法设定的检查点超时，如果设的太短，适当改长一点。另外就是考虑发生了反压或数据倾斜，或者barrier对齐太慢。

2. Flink的一些总结

2.1 状态后端的对比

2.2 反压监控参数

2.3 KeyedState和OperatorState的区别

2.4 Checkpoint文件说明

  • drwxr-xr-x 4 abc 74715970 128B Sep 23 03:19 job_127b2b84f80b368b8edfe02b2762d10d_op_KeyedProcessOperator_0d49016af99997646695a030f69aa7ee__1_1__uuid_65b50444-5857-4940-9f8c-77326cc79279/db
  • drwxr-xr-x 4 abc 74715970 128B Sep 23 03:20 job_127b2b84f80b368b8edfe02b2762d10d_op_StreamFlatMap_11f49afc24b1cce91c7169b1e5140284__1_1__uuid_19b333d3-3278-4e51-93c8-ac6c3608507c/db

3. Flink State最佳实践

3.1 Operator State使用建议

  • 慎重使用长list。

由于operator state没有key group的概念，所以为了实现改并发恢复的功能，需要对operator state中的每一个序列化后的元素存储一个位置偏移offset，也就是构成了上图红框中的offset数组。

那么如果你的operator state中的list长度达到一定规模时，这个offset数组就可能会有几十MB的规模，关键这个数组是会返回给job master，当operator的并发数目很大时，很容易触发job master的内存超用问题。我们遇到过用户把operator state当做黑名单存储，结果这个黑名单规模很大，导致一旦开始执行checkpoint，job master就会因为收到task发来的“巨大”的offset数组，而内存不断增长直到超用无法正常响应。

3.1.1 正确使用UnionListState

  • union list state目前被广泛使用在kafka connector中，不过可能用户日常开发中较少遇到，他的语义是从检查点恢复之后每个并发task内拿到的是原先所有operator上的state，如下图所示：

3.2 Keyed State使用建议

3.2.1 如何正确清空当前的state

3.2.2 KeyedState使用建议



3.2.3 ValueState大状态内存优化

  • 使用ValueState+HashMapStateBacked内存过大时，建议从以下分析：

3.2.3.1 ValueState容量缩减

3.2.3.2 ValueState数量减少

3.2.4 RawState使用建议

3.2.5 RocksDB使用建议

3.2.6 Checkpoint使用建议





end