Flink源码:从KeyGroup到Rescale

通过阅读本文你能 get 到以下点&＃xff1a;

• KeyGroup、KeyGroupRange 介绍
• maxParallelism 介绍及采坑记
• 数据如何映射到每个 subtask 上&＃xff1f;
• 任务改并发时&＃xff0c;KeyGroup rescale 的过程

一、 KeyGroup、KeyGroupRange 介绍

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

每个 Flink 的 subtask 负责一部分相邻 KeyGroup 的数据&＃xff0c;即一个 KeyGroupRange 的数据&＃xff0c;有个 start 和 end&＃xff08;这里是闭区间&＃xff09;。

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

二、 maxParallelism 介绍及采坑记

1、最大并行度的概念

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

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

2、maxParallelism 到底是多少呢&＃xff1f;

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

/*** Sets the maximum degree of parallelism defined for the program. The upper limit (inclusive)* is Short.MAX_VALUE.**

The maximum degree of parallelism specifies the upper limit for dynamic scaling. It also* defines the number of key groups used for partitioned state.** &＃64;param maxParallelism Maximum degree of parallelism to be used for the program.,* with 0 public StreamExecutionEnvironment setMaxParallelism(int maxParallelism) {Preconditions.checkArgument(maxParallelism > 0 &&maxParallelism <&＃61; KeyGroupRangeAssignment.UPPER_BOUND_MAX_PARALLELISM,"maxParallelism is out of bounds 0 &＃43;KeyGroupRangeAssignment.UPPER_BOUND_MAX_PARALLELISM &＃43; ". Found: " &＃43; maxParallelism);config.setMaxParallelism(maxParallelism);return this;}


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

/**根据算子的并行度计算 maxParallelism* 计算规则&＃xff1a;* 1. 将算子并行度 * 1.5 后&＃xff0c;向上取整到 2 的 n 次幂* 2. 跟 DEFAULT_LOWER_BOUND_MAX_PARALLELISM 相比&＃xff0c;取 max* 3. 跟 UPPER_BOUND_MAX_PARALLELISM 相比&＃xff0c;取 min*/
public static int computeDefaultMaxParallelism(int operatorParallelism) {checkParallelismPreconditions(operatorParallelism);return Math.min(Math.max(MathUtils.roundUpToPowerOfTwo(operatorParallelism &＃43; (operatorParallelism / 2)),DEFAULT_LOWER_BOUND_MAX_PARALLELISM),UPPER_BOUND_MAX_PARALLELISM);
}


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

最小值 DEFAULT_LOWER_BOUND_MAX_PARALLELISM 是 2 的 7 次方 &＃61; 128。

最大值 UPPER_BOUND_MAX_PARALLELISM 是 2 的 15 次方 &＃61; 32768。

即&＃xff1a;Flink 自动生成的 maxParallelism 介于 128 和 32768 之间。

采坑记

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

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

三、每个 key 应该分配到哪个 subtask 上运行&＃xff1f;

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

• 第一步&＃xff1a;根据 key 计算其对应哪个 KeyGroup
• 第二步&＃xff1a;计算 KeyGroup 属于哪个并行度

第一步&＃xff1a;根据 key 计算其对应哪个 KeyGroup

computeKeyGroupForKeyHash 源码如下所示&＃xff1a;

/*** Assigns the given key to a key-group index.** &＃64;param keyHash the hash of the key to assign* &＃64;param maxParallelism the maximum supported parallelism, aka the number of key-groups.* &＃64;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 后&＃xff0c;对 maxParallelism 进行求余&＃xff0c;就是对应的 KeyGroupIndex。

第二步&＃xff1a;计算 KeyGroup 属于哪个并行度

computeOperatorIndexForKeyGroup 源码如下所示&＃xff1a;

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


示例

假如 maxParallelism 为 50&＃xff0c;parallelism 为 10&＃xff0c;那么数据是如何分布的&＃xff1f;

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

subtask 与 KeyGroupId 对应关系&＃xff1a;

• 0~4 号 KeyGroup 位于第 0 个 subtask。即&＃xff1a;subtask0 处理 KeyGroupRange(0,4 ) 的数据
• 5~9 号 KeyGroup 位于第 1 个 subtask。即&＃xff1a;subtask1 处理 KeyGroupRange(5,9 ) 的数据
• 10~14 号 KeyGroup 位于第 2 个 subtask。即&＃xff1a;subtask2 处理 KeyGroupRange(10,14 ) 的数据
• 15~19 号 KeyGroup 位于第 3 个 subtask。即&＃xff1a;subtask3 处理 KeyGroupRange(15,19 ) 的数据
• 。。。以此类推

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

计算某个并行度上负载哪些 KeyGroup&＃xff1f;

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

在 KeyGroupRangeAssignment 类中有 computeKeyGroupRangeForOperatorIndex 方法可以完成这个操作&＃xff1a;

// 根据 maxParallelism, parallelism 计算 operatorIndex 对应的 subtask 负责哪个范围的 KeyGroup
public static KeyGroupRange computeKeyGroupRangeForOperatorIndex(int maxParallelism,int parallelism,int operatorIndex) {checkParallelismPreconditions(parallelism);checkParallelismPreconditions(maxParallelism);Preconditions.checkArgument(maxParallelism >&＃61; parallelism,"Maximum parallelism must not be smaller than parallelism.");int start &＃61; ((operatorIndex * maxParallelism &＃43; parallelism - 1) / parallelism);int end &＃61; ((operatorIndex &＃43; 1) * maxParallelism - 1) / parallelism;return new KeyGroupRange(start, end);
}


四、 Rescale 过程

如下图所示是 Flink 依赖 KeyGroup 修改并发的 Rescale 过程&＃xff08;并发度从 3 改成 4&＃xff09;&＃xff1a;

由图中可得知 key 的范围是 0~19&＃xff0c; maxParallelism &＃61; 10。

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

1->{1,11} 表示 key 为 1 和 11 的数据&＃xff0c;对应的 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&＃xff0c;一旦 maxParallelism 变了&＃xff0c;说明 KeyGroup 的分组完全变了&＃xff0c;而 KeyedState 恢复是以 KeyGroup 为最小单元的&＃xff0c;所以 maxParallelism 改变后&＃xff0c;任务将无法恢复。在 Checkpoint 恢复过程中也会对新旧 Job 的 maxParallelism 进行检查匹配&＃xff0c;如果某个算子的 maxParallelism 变了&＃xff0c;则任务将不能恢复。

五、总结

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

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