You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user-zh@flink.apache.org by "junjie.miao@goupwith.com" <ju...@goupwith.com> on 2022/09/09 09:58:31 UTC
回复: Re: Key group is not in KeyGroupRange
Integer[] rebalanceKeys = createRebalanceKeys(parallelism);
int rebalanceKeyIndex = new Random().nextInt(parallelism);
Integer key = rebalanceKeys[rebalanceKeyIndex];
/**
* 构建均衡 KEY 数组
*
* @param parallelism 并行度
* @return
*/
public static Integer[] createRebalanceKeys(int parallelism) {
HashMap<Integer, LinkedHashSet<Integer>> groupRanges = new HashMap<>();
int maxParallelism = KeyGroupRangeAssignment.computeDefaultMaxParallelism(parallelism);
// 构造多个 key 用于生成足够的 groupRanges
int maxRandomKey = parallelism * 10;
for (int randomKey = 0; randomKey < maxRandomKey; randomKey++) {
int subtaskIndex = KeyGroupRangeAssignment.assignKeyToParallelOperator(randomKey, maxParallelism, parallelism);
LinkedHashSet<Integer> randomKeys = groupRanges.computeIfAbsent(subtaskIndex, k -> new LinkedHashSet<>());
randomKeys.add(randomKey);
}
Integer[] result = new Integer[parallelism];
for (int i = 0; i < parallelism; i++) {
LinkedHashSet<Integer> ranges = groupRanges.get(i);
if (ranges == null || ranges.isEmpty()) {
throw new RuntimeException("create rebalance keys error");
}
result[i] = ranges.stream().findFirst().get();
}
return result;
}
发件人： junjie.miao@goupwith.com
发送时间： 2022-09-09 17:52
收件人： user-zh
主题： Re: Re: Key group is not in KeyGroupRange
key selector中使用random.nextInt(parallelism) 有时会报错
From: yue ma
Date: 2022-09-09 17:41
To: user-zh
Subject: Re: Key group is not in KeyGroupRange
你好，可以看一下使用的 key selector 是否稳定，key 是否会变化。
junjie.miao@goupwith.com <ju...@goupwith.com> 于2022年9月9日周五 17:35写道：
> hi:
> 本人遇到了这个报错：
> Key group 51 is not in KeyGroupRange{startKeyGroup=64, endKeyGroup=127}.
> Unless you're directly using low level state access APIs, this is most
> likely caused by non-deterministic shuffle key (hashCode and equals
> implementation).
>
> 这个报错出现的莫名其妙，想了解这个错误是在什么样的情况下才会导致这个报错的？
> 谢谢！！
>
>
Re: Re: Key group is not in KeyGroupRange
Posted by yidan zhao <hi...@gmail.com>.
你发的代码就是原因，keySelector对于同一个输入数据，输出结果必须一致，不能不同。
如果需要实现类似效果，可以先使用 flatMap 生成 random key，然后将 key 存储到一个字段，比如就叫做 key，然后
keyBy("key") 这样可以。
junjie.miao@goupwith.com <ju...@goupwith.com> 于2022年9月9日周五 17:59写道：
>
>
> Integer[] rebalanceKeys = createRebalanceKeys(parallelism);
> int rebalanceKeyIndex = new Random().nextInt(parallelism);
> Integer key = rebalanceKeys[rebalanceKeyIndex];
>
> /**
> * 构建均衡 KEY 数组
> *
> * @param parallelism 并行度
> * @return
> */
> public static Integer[] createRebalanceKeys(int parallelism) {
> HashMap<Integer, LinkedHashSet<Integer>> groupRanges = new HashMap<>();
> int maxParallelism = KeyGroupRangeAssignment.computeDefaultMaxParallelism(parallelism);
> // 构造多个 key 用于生成足够的 groupRanges
> int maxRandomKey = parallelism * 10;
> for (int randomKey = 0; randomKey < maxRandomKey; randomKey++) {
> int subtaskIndex = KeyGroupRangeAssignment.assignKeyToParallelOperator(randomKey, maxParallelism, parallelism);
> LinkedHashSet<Integer> randomKeys = groupRanges.computeIfAbsent(subtaskIndex, k -> new LinkedHashSet<>());
> randomKeys.add(randomKey);
> }
>
> Integer[] result = new Integer[parallelism];
> for (int i = 0; i < parallelism; i++) {
> LinkedHashSet<Integer> ranges = groupRanges.get(i);
> if (ranges == null || ranges.isEmpty()) {
> throw new RuntimeException("create rebalance keys error");
> }
> result[i] = ranges.stream().findFirst().get();
> }
> return result;
> }
>
>
> 发件人： junjie.miao@goupwith.com
> 发送时间： 2022-09-09 17:52
> 收件人： user-zh
> 主题： Re: Re: Key group is not in KeyGroupRange
> key selector中使用random.nextInt(parallelism) 有时会报错
>
> From: yue ma
> Date: 2022-09-09 17:41
> To: user-zh
> Subject: Re: Key group is not in KeyGroupRange
> 你好，可以看一下使用的 key selector 是否稳定，key 是否会变化。
> junjie.miao@goupwith.com <ju...@goupwith.com> 于2022年9月9日周五 17:35写道：
> > hi:
> > 本人遇到了这个报错：
> > Key group 51 is not in KeyGroupRange{startKeyGroup=64, endKeyGroup=127}.
> > Unless you're directly using low level state access APIs, this is most
> > likely caused by non-deterministic shuffle key (hashCode and equals
> > implementation).
> >
> > 这个报错出现的莫名其妙，想了解这个错误是在什么样的情况下才会导致这个报错的？
> > 谢谢！！
> >
> >