Flink 源码:从 KeyGroup 到 Rescale

通过阅读本文你能 get 到以下点:

  • KeyGroup、KeyGroupRange 介绍

  • maxParallelism 介绍及采坑记

  • 数据如何映射到每个 subtask 上?

  • 任务改并发时,KeyGroup rescale 的过程

一、 KeyGroup、KeyGroupRange 介绍

Flink 中 KeyedState 恢复时,是按照 KeyGroup 为最小单元恢复的,每个 KeyGroup 负责一部分 key 的数据。这里的 key 指的就是 Flink 中 keyBy 中提取的 key。

每个 Flink 的 subtask 负责一部分相邻 KeyGroup 的数据,即一个 KeyGroupRange 的数据,有个 start 和 end(这里是闭区间)。

看到这里可能有点蒙,没关系后面有例子帮助读者理解这两个概念。

二、 maxParallelism 介绍及采坑记

最大并行度的概念

maxParallelism 表示当前算子设置的 maxParallelism,而不是 Flink 任务的并行度。maxParallelism 为 KeyGroup 的个数。

当设置算子的并行度大于 maxParallelism 时,有些并行度就分配不到 KeyGroup,此时 Flink 任务是无法从 Checkpoint 处恢复的。

maxParallelism 到底是多少呢?

如果设置了,就是设定的值。当然设置了,也需要检测合法性。如下图所示,Flink 要求 maxParallelism 应该介于 1 到 Short.MAX_VALUE 之间。

如果没有设置,则 Flink 引擎会自动通过 KeyGroupRangeAssignment 类的 computeDefaultMaxParallelism 方法计算得出,computeDefaultMaxParallelism 源码如下所示:

/**根据算子的并行度计算 maxParallelism
 * 计算规则:
 * 1. 将算子并行度 * 1.5 后,向上取整到 2 的 n 次幂
 * 2. 跟 DEFAULT_LOWER_BOUND_MAX_PARALLELISM 相比,取 max
 * 3. 跟 UPPER_BOUND_MAX_PARALLELISM 相比,取 min
 */
public static int computeDefaultMaxParallelism(int operatorParallelism) {

 checkParallelismPreconditions(operatorParallelism);

 return Math.min(
   Math.max(
     MathUtils.roundUpToPowerOfTwo(operatorParallelism + (operatorParallelism / 2)),
     DEFAULT_LOWER_BOUND_MAX_PARALLELISM),
   UPPER_BOUND_MAX_PARALLELISM);
}

computeDefaultMaxParallelism 会根据算子的并行度计算 maxParallelism,计算规则:将算子并行度 * 1.5 后,向上取整到 2 的 n 次幂,同时保证计算的结果在最小值和最大值之间。

最小值 DEFAULT_LOWER_BOUND_MAX_PARALLELISM 是 2 的 7 次方 = 128。

最大值 UPPER_BOUND_MAX_PARALLELISM 是 2 的 15 次方 = 32768。

即:Flink 自动生成的 maxParallelism 介于 128 和 32768 之间。

新开发的 Job 业务数据量较小,所以初期设置的并行度也会很小。同时没有给每个 Job 主动设置 maxParallelism,根据上面的规则,Flink 自动生成的 maxParallelism 为 128,后期随着业务数据量暴涨,当 Job 的并发数调大 128 以上时,发现 Job 无法从 Checkpoint 或 Savepoint 中恢复了,这就是所谓的 "并发调不上去了"。当然可以选择不从状态恢复,选择直接启动的方式去启动任务。但是有些 Flink 任务对状态是强依赖的,即:必须从 State 中恢复,对于这样的 Job 就不好办了。

所以按照开发规范,应该结合业务场景主动为每个 Job 设置合理的 maxParallelism,防止出现类似情况。

三、每个 key 应该分配到哪个 subtask 上运行?

根据 key 计算其对应的 subtaskIndex,即应该分配给哪个 subtask 运行,计算过程包括以下两步,源码都在相应的 KeyGroupRangeAssignment 类中:

  • 第一步:根据 key 计算其对应哪个 KeyGroup

  • 第二步:计算 KeyGroup 属于哪个并行度

第一步:根据 key 计算其对应哪个 KeyGroup

computeKeyGroupForKeyHash 源码如下所示:

/**
 * Assigns the given key to a key-group index.
 *
 * @param keyHash the hash of the key to assign
 * @param maxParallelism the maximum supported parallelism, aka the number of key-groups.
 * @return the key-group to which the given key is assigned
 * 根据 Key 的 hash 值来计算其对应的 KeyGroup 的 index
 */
public static int computeKeyGroupForKeyHash(int keyHash, int maxParallelism) {
 return MathUtils.murmurHash(keyHash) % maxParallelism;
}

根据 Key 的 hash 值进行 murmurHash 后,对 maxParallelism 进行求余,就是对应的 KeyGroupIndex。

第二步:计算 KeyGroup 属于哪个并行度

computeOperatorIndexForKeyGroup 源码如下所示:

// 根据 maxParallelism、算子的并行度 parallelism 和 keyGroupId,
// 计算 keyGroupId 对应的 subtask 的 index
public static int computeOperatorIndexForKeyGroup(int maxParallelism, 
                                                  int parallelism, int keyGroupId) {
 return keyGroupId * parallelism / maxParallelism;
}

假如 maxParallelism 为 50,parallelism 为 10,那么数据是如何分布的?

MathUtils.murmurHash(key.hashCode()) % maxParallelism:所有 key 的 hashCode 通过 Murmurhash 对 50 求余得到的范围为 0~49,也就是说:总共有 keyGroupId 为 0~49 的这 50 个 KeyGroup。

subtask 与 KeyGroupId 对应关系:

  • 0~4 号 KeyGroup 位于第 0 个 subtask。即:subtask0 处理 KeyGroupRange(0,4 ) 的数据

  • 5~9 号 KeyGroup 位于第 1 个 subtask。即:subtask1 处理 KeyGroupRange(5,9 ) 的数据

  • 10~14 号 KeyGroup 位于第 2 个 subtask。即:subtask2 处理 KeyGroupRange(10,14 ) 的数据

  • 15~19 号 KeyGroup 位于第 3 个 subtask。即:subtask3 处理 KeyGroupRange(15,19 ) 的数据

  • 。。。以此类推

这里我们看到了每个 subtask 对应一个 KeyGroupRange 的数据,且是闭区间。

计算某个并行度上负载哪些 KeyGroup?

计算某个并行度上负载哪些 KeyGroup?等价于求某个 subtask 负载的 KeyGroupRange。

在 KeyGroupRangeAssignment 类中有 computeKeyGroupRangeForOperatorIndex 方法可以完成这个操作:

// 根据 maxParallelism, parallelism 计算 operatorIndex 对应的 subtask 负责哪个范围的 KeyGroup
public static KeyGroupRange computeKeyGroupRangeForOperatorIndex(
 int maxParallelism,
 int parallelism,
 int operatorIndex) {

 checkParallelismPreconditions(parallelism);
 checkParallelismPreconditions(maxParallelism);

 Preconditions.checkArgument(maxParallelism >= parallelism,
  "Maximum parallelism must not be smaller than parallelism.");

 int start = ((operatorIndex * maxParallelism + parallelism - 1) / parallelism);
 int end = ((operatorIndex + 1) * maxParallelism - 1) / parallelism;
 return new KeyGroupRange(start, end);
}

四、 Rescale 过程

如下图所示是 Flink 依赖 KeyGroup 修改并发的 Rescale 过程(并发度从 3 改成 4):

KeyGroup Rescale 过程图示

由图中可得知 key 的范围是 0~19, maxParallelism = 10。

0->{0,10} 表示 key 为 0 和 10 的数据,对应的 KeyGroupId 为 0。

1->{1,11} 表示 key 为 1 和 11 的数据,对应的 KeyGroupId 为 1。

以此类推。。。

修改并发前的映射关系

并发度是 3:

  • subtask0 负责 KeyGroupRange(0,3)

  • Subtask1 负责 KeyGroupRange(4,6)

  • Subtask2 负责 KeyGroupRange(7,9)

修改并发后的映射关系

并发度是 4:

  • subtask0 负责 KeyGroupRange(0,2)

  • Subtask1 负责 KeyGroupRange(3,4)

  • Subtask2 负责 KeyGroupRange(5,7)

  • Subtask3 负责 KeyGroupRange(8,9)

maxParallelism 修改则任务不能恢复

KeyGroup 的数量为 maxParallelism,一旦 maxParallelism 变了,说明 KeyGroup 的分组完全变了,而 KeyedState 恢复是以 KeyGroup 为最小单元的,所以 maxParallelism 改变后,任务将无法恢复。在 Checkpoint 恢复过程中也会对新旧 Job 的 maxParallelism 进行检查匹配,如果某个算子的 maxParallelism 变了,则任务将不能恢复。

五、总结

本文主要介绍了 KeyGroup、KeyGroupRange 和 maxParallelism 的一些概念,及他们之间的关系。最后讲述了改并发的情况状态的 Rescale 流程。其实在 Flink 内部不只是状态恢复时需要用到 KeyGroup,数据 keyBy 后进行 shuffle 数据传输时也需要按照 KeyGroup 的规则来将分配数据,将数据分发到对应的 subtask 上。

本文比较简单,主要是为后续 State 恢复流程做一个铺垫。

我来评几句
登录后评论

已发表评论数()

相关站点

+订阅
热门文章