为什么 flink 作业的 maxparallelism 不能在不丢失状态的情况下更新？

Question

我刚刚读到 Flink 作业的最大并行度（由 setMaxParallelism 定义）不能在不丢失状态的情况下更改。这让我有点吃惊，不难想象一个人开始运行作业，却发现负载最终比预期大 10 倍（或者代码的效率低于预期）导致希望增加并行度。

除了一些对关键组的引用之外，我找不到很多原因。我发现的最有形的陈述here：

在扩展作业时不能改变最大并行度，因为它会破坏键到键组的映射。

但是，这仍然给我留下了一些问题：

为什么让工作改变其最大并行度很难/不可能？

---

基于上述，我想到了以下概念性解决方案：

1. 在状态中，跟踪上次使用的最大并行度
2. 开始作业时，指明所需的最大并行度
3. 鉴于这两个设置都是已知的，应该可以推断出映射需要如何更改才能保持初始有效。
4. 如果需要，可以使用新的 maxparallelism 基于旧状态定义新状态，以“适应”新作业。

我并不是说这个概念性解决方案是理想的，或者说实施起来很简单。我只是想知道最大并行度的非常严格的性质是否还有更多。并试图理解这是否只是“这种灵活性尚未实现”或“这与 Flink 的本质相悖，以至于人们不应该想要它”的问题。

Answer 1

通过以密钥组的数量为模计算密钥的哈希值，将每个密钥分配给一个密钥组。因此，更改键组的数量会影响键组的键分配。每个任务管理器负责一个或多个键组，因此键组的数量与最大并行度相同。

这个数字很难更改的原因是它有效地融入了状态快照（检查点和保存点）。这些快照按键组索引，因此在系统启动时，每个任务管理器都可以有效地加载他们需要的状态。

随着键组数量的增加，内存中的数据结构会显着扩展，这就是为什么最大并行度不会默认为某个相当大的值（默认值为 128）。

State Processor API 可用于重写状态快照，如果您需要更改密钥组的数量，或在状态后端之间迁移。