最近在做大数据量的实时数据迁徙, 频繁应用到了 keyby hash 去平衡数据, 然而却发现 subtask 执行的数据量不是很平衡, 导致 checkpoint 频繁超时, 于是开始寻找解决办法.
问题背景
应用 keyby 进行分区, 自定义 KeySelector
, 进行hash% 并行度
来进行分区, 比方应用的并行度是 8, 最初会失去分区 key
0, 1, 2, 3, 4, 5, 6, 7
run 起我的项目后, 关上监督后盾, 发现 8 个 subtask 中有一个 task 没有数据, 另一个 task 会有双倍的数据.
起因
具体参考官网文档
起因很简略, flink 无奈预估你会有多少 key, 所以会基于最大并行度 (默认 128) 进行一个 key 分组, 在这个范畴内的才会调配到 task 中.
以下是相干代码
public static int computeKeyGroupForKeyHash(int keyHash, int maxParallelism) {return MathUtils.murmurHash(keyHash) % maxParallelism;
}
那么咱们对下面本人的 key, 去运行这段代码, 会失去以下后果
0 86
1 54
2 27
3 33
4 4
5 79
6 19
7 115
咱们是 8 个并行, 相当于每个 subtask 占有 16 个 key, 会失去以下分组:
0~15 4
16~31 19 27
32~47 33
48~63 54
64~79 79
80~95 86
96~111
112~127 115
会发现有个分区的确取得了两个 key, 而一个分区轮空.
解
将 6 换成 murmurhash 后在 96~111 中的 key, 比方 6666
(hash 为 106)
重启之后调配不均的问题解决.
終わり。