在 Flink 中设置适当的运算符并行度的直觉
Intuition for setting appropriate parallelism of operators in Flink
我的问题是关于了解在固定集群设置中的 flink 作业中的运算符的并行性的良好选择。假设，我们有一个包含 map 和 reduce 类型运算符的 flink 作业 DAG，它们之间有流水线边缘（没有阻塞边缘）。 DAG示例如下：
Scan -> Keyword Search -> Aggregation
假设一个固定大小的 M 机器集群，每个机器具有 C 个核心，DAG 是集群上唯一 运行 的工作流。 Flink 允许用户为各个算子设置并行度。我通常为每个运算符设置 M*C 并行度。但从性能角度（例如执行时间）来看，这是最佳选择吗？我们能否利用算子的属性做出更好的选择？例如，如果我们知道 aggregation 更昂贵，我们是否应该将 M*C 并行度仅分配给 aggregation 运算符并减少其他运算符的并行度？希望这也能减少背压的可能性。
我不是在寻找能给我“最佳”并行性的合适公式。我只是在寻找某种可以用来做出决定的 intuition/guideline/ideas 。令人惊讶的是，我找不到太多关于这个主题的文献。
注意：我知道最近Flink中的动态缩放反应模式。但我的问题是关于只有一个工作流 运行ning 的固定集群，这意味着动态缩放不相关。我查看了 this 问题，但没有得到答案。
我对此的看法有些不同。在我看来，有两个关键问题需要考虑：
(1) 我要让插槽保持统一吗？或者换句话说，是否每个槽都有每个任务的实例，或者我想调整特定任务的并行度？
(2) 每个插槽有多少个内核？
我对 (1) 的默认回答是“保持统一”。我还没有看到很多情况证明调整单个操作符（或任务）的并行性是值得的。
如果改变并行度意味着破坏操作符链，通常会适得其反。在不寻常的情况下无论如何都可以在洗牌的地方进行，但总的来说我不明白这一点。由于某些槽将包含每个运算符的实例，并且槽都是统一的，为什么分配给它们的任务较少的槽会有帮助？ （在这里，我假设您对设置插槽共享组的麻烦不感兴趣，这当然是可以做到的。）从操作的角度来看，沿着这条路走下去会使事情变得更加复杂，而且收效甚微。在我看来，最好在其他地方进行优化（例如序列化）。
至于每个插槽的内核数，许多作业受益于每个插槽 2 个内核，而对于一些具有大量任务的复杂作业，您可能希望更高。因此，我认为对于简单的 ETL 作业而言 M*C 的总体并行性，对于执行更激烈的作业的 M*C/2（或更低）。
举例说明极端情况：
一个简单的 ETL 作业可能类似于
source -> map -> sink
其中所有连接都是转发连接。由于只有一个任务，并且因为 Flink 每个任务只使用一个线程，在这种情况下我们每个槽只使用一个线程。因此，为每个插槽分配一个以上的内核完全是一种浪费。无论如何，任务可能 i/o 绑定。
在另一个极端，我见过涉及约 30 个连接的工作，一个或多个 ML 模型的评估，以及窗口聚合等。您当然需要多个 CPU 核心处理每个像这样的工作的平行部分（就此而言，超过两个）。
通常，大部分 CPU 工作都用于序列化和反序列化，尤其是 RocksDB。我会尝试弄清楚，对于每个事件，涉及多少 RocksDB 状态访问、keyBy 和重新平衡——并提供足够的核心以使所有这些 ser/de 可以同时发生（如果您关心最大化吞吐量）。对于最简单的工作，一个核心就可以跟上。到您进行诸如窗口连接之类的操作时，您可能已经突破了一个核心可以跟上的极限——这取决于您的源和接收器的运行速度，以及您在不浪费资源方面的谨慎程度。
我的问题是关于了解在固定集群设置中的 flink 作业中的运算符的并行性的良好选择。假设，我们有一个包含 map 和 reduce 类型运算符的 flink 作业 DAG，它们之间有流水线边缘（没有阻塞边缘）。 DAG示例如下：
Scan -> Keyword Search -> Aggregation
假设一个固定大小的 M 机器集群，每个机器具有 C 个核心，DAG 是集群上唯一 运行 的工作流。 Flink 允许用户为各个算子设置并行度。我通常为每个运算符设置 M*C 并行度。但从性能角度（例如执行时间）来看，这是最佳选择吗？我们能否利用算子的属性做出更好的选择？例如，如果我们知道 aggregation 更昂贵，我们是否应该将 M*C 并行度仅分配给 aggregation 运算符并减少其他运算符的并行度？希望这也能减少背压的可能性。
我不是在寻找能给我“最佳”并行性的合适公式。我只是在寻找某种可以用来做出决定的 intuition/guideline/ideas 。令人惊讶的是，我找不到太多关于这个主题的文献。
注意：我知道最近Flink中的动态缩放反应模式。但我的问题是关于只有一个工作流 运行ning 的固定集群，这意味着动态缩放不相关。我查看了 this 问题，但没有得到答案。
我对此的看法有些不同。在我看来，有两个关键问题需要考虑：
(1) 我要让插槽保持统一吗？或者换句话说，是否每个槽都有每个任务的实例，或者我想调整特定任务的并行度？
(2) 每个插槽有多少个内核？
我对 (1) 的默认回答是“保持统一”。我还没有看到很多情况证明调整单个操作符（或任务）的并行性是值得的。
如果改变并行度意味着破坏操作符链，通常会适得其反。在不寻常的情况下无论如何都可以在洗牌的地方进行，但总的来说我不明白这一点。由于某些槽将包含每个运算符的实例，并且槽都是统一的，为什么分配给它们的任务较少的槽会有帮助？ （在这里，我假设您对设置插槽共享组的麻烦不感兴趣，这当然是可以做到的。）从操作的角度来看，沿着这条路走下去会使事情变得更加复杂，而且收效甚微。在我看来，最好在其他地方进行优化（例如序列化）。
至于每个插槽的内核数，许多作业受益于每个插槽 2 个内核，而对于一些具有大量任务的复杂作业，您可能希望更高。因此，我认为对于简单的 ETL 作业而言 M*C 的总体并行性，对于执行更激烈的作业的 M*C/2（或更低）。
举例说明极端情况：
一个简单的 ETL 作业可能类似于
source -> map -> sink
其中所有连接都是转发连接。由于只有一个任务，并且因为 Flink 每个任务只使用一个线程，在这种情况下我们每个槽只使用一个线程。因此，为每个插槽分配一个以上的内核完全是一种浪费。无论如何，任务可能 i/o 绑定。
在另一个极端，我见过涉及约 30 个连接的工作，一个或多个 ML 模型的评估，以及窗口聚合等。您当然需要多个 CPU 核心处理每个像这样的工作的平行部分（就此而言，超过两个）。
通常，大部分 CPU 工作都用于序列化和反序列化，尤其是 RocksDB。我会尝试弄清楚，对于每个事件，涉及多少 RocksDB 状态访问、keyBy 和重新平衡——并提供足够的核心以使所有这些 ser/de 可以同时发生（如果您关心最大化吞吐量）。对于最简单的工作，一个核心就可以跟上。到您进行诸如窗口连接之类的操作时，您可能已经突破了一个核心可以跟上的极限——这取决于您的源和接收器的运行速度，以及您在不浪费资源方面的谨慎程度。