如何计算用于合并的最佳分区数？

Question

所以，我知道一般来说，在以下情况下应该使用 coalesce（）：

由于过滤器或其他可能导致减少原始数据集（RDD、DF）的操作，分区数量减少。coalesce（）对于在过滤大型数据集后更有效地运行操作很有用。

我也知道它比< code>repartition更便宜，因为它通过仅在必要时移动数据来减少洗牌。我的问题是如何定义< code>coalesce采用的参数(< code > idealpartionno )。我正在做一个项目，这个项目是另一个工程师交给我的，他用下面的计算来计算这个参数的值。

// DEFINE OPTIMAL PARTITION NUMBER
implicit val NO_OF_EXECUTOR_INSTANCES = sc.getConf.getInt("spark.executor.instances", 5)
implicit val NO_OF_EXECUTOR_CORES = sc.getConf.getInt("spark.executor.cores", 2)

val idealPartionionNo = NO_OF_EXECUTOR_INSTANCES * NO_OF_EXECUTOR_CORES * REPARTITION_FACTOR

然后将其用于分区器对象：

val partitioner = new HashPartitioner(idealPartionionNo)

但也用于：

RDD.filter(x=>x._3<30).coalesce(idealPartionionNo)

这是正确的方法吗？< code > idealpartionno 值计算背后的主要思想是什么？什么是< code >重新分配因子？我通常如何定义它？

此外，由于YARN负责实时识别可用的执行器，是否有一种方法可以实时获取该编号（available_EXECUTOR_INSTANCES），并将其用于计算理想PartitionNo（即将 替换为

理想情况下，以下形式的一些实际例子:

• 这是一个数据集（大小）
• 下面是RDD/DF的一些转换和可能的重用
• 这里是您应该重新分区/合并的地方
• 假设您有n个执行器，其中m个核，分区因子等于

然后：

• 理想的分区数应该是==

另外，如果你能给我推荐一个很好的博客来解释这些，我将不胜感激。

Answer 1

正如其他人所回答的，没有公式可以计算你的要求。也就是说，你可以对第一部分进行有根据的猜测，然后随着时间的推移对其进行微调。

第一步是确保您有足够的分区。如果您没有_OF_EXECUTOR_INSTANCES个执行程序，并且每个执行程序没有_OF_EXECUTOR_CORES个核心，那么您可以同时处理NO _ OF _ EXECUTOR _ INSTANCES * NO _ OF _ EXECUTOR _ CORES个分区(每个分区将处理特定实例的特定核心)。也就是说，这是假设所有东西都在内核之间平均分配，并且所有东西都需要完全相同的时间来处理。这种情况很少发生。由于本地性(例如，数据需要来自不同的节点)或者仅仅是因为它们不平衡(例如，如果您有按根域分区的数据，那么包括google在内的分区可能会非常大)，其中一些很可能会在其他分区之前完成。这就是REPARTITION_FACTOR发挥作用的地方。这个想法是，我们“超额预定”每个内核，因此如果一个内核完成得非常快，一个内核完成得很慢，我们可以选择在它们之间分配任务。系数2-3通常是个好主意。

现在让我们看一下单个分区的大小。假设您的整个数据大小为 X MB，并且您有 N 个分区。每个分区的平均 X/N MB 数。如果 N 相对于 X 很大，那么您可能具有非常小的平均分区大小（例如几 KB）。在这种情况下，降低 N 通常是一个好主意，因为管理每个分区的开销变得太高。另一方面，如果大小非常大（例如几GB），那么您需要同时保存大量数据，这会导致垃圾回收，高内存使用率等问题。

最佳大小是一个很好的问题，但通常人们似乎更喜欢100-1000MB的分区，但事实上，几十MB的分区可能也很好。

您应该注意的另一件事是，在计算分区如何变化时。例如，假设您从1000个分区开始，每个分区100MB，然后过滤数据，使每个分区变为1K，那么您可能应该合并。当您进行分组或加入时，可能会发生类似的问题。在这种情况下，分区的大小和分区的数量都会发生变化，可能会达到不期望的大小。

Answer 2

您的问题是正确的，但Spark分区优化完全取决于您正在运行的计算。你需要有一个很好的理由来重新划分/合并；如果您只是计算一个RDD（即使它有大量稀疏的分区），那么任何重新分区/合并步骤都会让您慢下来。

重新分区（n）（与coalesce（n，shuffle=true）和 的区别与执行模型有关。shuffle模型获取原始RDD中的每个分区，将其数据随机发送给所有执行器，并生成具有新（更小或更大）分区数的RDD。无洗牌模型创建一个新的RDD，将多个分区作为一个任务加载。

让我们考虑这样的计算:

sc.textFile("massive_file.txt")
  .filter(sparseFilterFunction) // leaves only 0.1% of the lines
  .coalesce(numPartitions, shuffle = shuffle)

如果< code>shuffle为< code>true，则文本文件/过滤器计算发生在由< code>textFile中的缺省值给定的许多任务中，微小的过滤结果被打乱。如果< code>shuffle为< code>false，则任务总数最多为< code>numPartitions。

如果numPartitions为1，则差异非常明显。shuffle模型将并行处理和过滤数据，然后将0.1%的过滤结果发送给一个执行器，用于下游DAG操作。无混洗模型将从一开始就在一个核心上处理和过滤数据。

考虑您的下游操作。如果只使用一次该数据集，那么可能根本不需要重新分区。如果您要将过滤后的RDD保存以供以后使用（例如，保存到磁盘），那么请考虑上面的权衡。熟悉这些模型需要经验，当一个模型表现更好时，请尝试这两种模型，看看它们的表现如何！

Answer 3

在实践中，最佳分区数量更多地取决于您拥有的数据、您使用的转换和整体配置，而不是可用资源。

• 如果分区数太少，则会遇到长时间的 GC 暂停、不同类型的内存问题，最后是资源利用率欠佳的情况。
• 如果分区数太大，则维护成本很容易超过处理成本。此外，如果使用非分布式归约操作（如与 treeReduce 相比的 reduce），则大量分区会导致驱动程序上的负载增加。

您可以找到许多规则，这些规则建议超额订阅分区与内核数量（因子 2 或 3 似乎很常见）或将分区保持在一定大小，但这不考虑您自己的代码：

• 如果您分配了很多，则可以预期长时间的GC暂停，最好使用较小的分区。
• 如果某段代码很昂贵，那么您的随机播放成本可以通过更高的并发性来摊销。
• 如果您有筛选器，则可以根据谓词的判别能力调整分区数（如果您希望保留 5% 的数据和 99% 的数据，则会做出不同的决定）。

在我看来：

>

对于一次性作业，保持更高数量的分区以保持安全（慢比失败好）。

使用可重用作业，从保守配置开始，然后执行-监控-调整配置-重复。

不要尝试根据执行器或核心的数量使用固定数量的分区。首先了解您的数据和代码，然后调整配置以反映您的理解。

通常，确定集群表现出稳定行为的每个分区的原始数据量相对容易（根据我的经验，它在几百兆字节的范围内，这取决于您用于加载数据的格式、数据结构和配置）。这就是你要找的“神奇数字”。

总的来说，有些事情你必须记住:

• 分区数不一定反映数据分布。任何需要随机操作（*byKey，join，RDD.partitionBy，Dataset.repartition）都可能导致数据分布不均匀。始终监视作业是否存在严重数据偏差的症状。
• 分区数通常不是恒定的。具有多个依赖项（联合、coGroup、联接）的任何操作都可能影响分区数。