scala - 如何计算合并的最佳分区数？

Question

所以，我知道通常应该使用 coalesce()什么时候:

the number of partitions decreases due to a filter or some other operation that may result in reducing the original dataset (RDD, DF). coalesce() is useful for running operations more efficiently after filtering down a large dataset.

我也明白它比 repartition便宜因为它仅在必要时通过移动数据来减少洗牌。我的问题是如何定义 coalesce 的参数需要( idealPartionionNo)。我正在处理一个从另一位工程师传递给我的项目，他正在使用以下计算来计算该参数的值。

// 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

然后将其与 partitioner 一起使用目的:

val partitioner = new HashPartitioner(idealPartionionNo)

但也用于:

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

这是正确的方法吗？ idealPartionionNo 背后的主要思想是什么？值(value)计算？什么是 REPARTITION_FACTOR ?我通常如何定义它？

此外，由于 YARN 负责动态识别可用的执行程序，因此有一种方法可以即时获取该数字( AVAILABLE_EXECUTOR_INSTANCES )并将其用于计算 idealPartionionNo (即用 NO_OF_EXECUTOR_INSTANCES 替换 AVAILABLE_EXECUTOR_INSTANCES )？

理想情况下，表单的一些实际示例:

这是一个数据集(大小)；

这是 RDD/DF 的一些转换和可能的重用。

这是您应该重新分区/合并的地方。

假设您有 n执行人 m核心和等于 k 的分配因子

然后:

理想的分区数是 ==> ???

另外，如果您可以将我推荐给一个很好的博客来解释这些，我将非常感激。

Answer 1

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

如果分区数量太少，您将遇到长时间的 GC 暂停、不同类型的内存问题，最后是资源利用率不佳。

如果分区数量过多，则维护成本很容易超过处理成本。此外，如果您使用非分布式归约操作(如 reduce 与 treeReduce 相比)，大量分区会导致驱动程序负载更高。

您可以找到一些规则，这些规则建议与内核数量相比过度订阅分区(因子 2 或 3 似乎很常见)或将分区保持在一定大小，但这并没有考虑到您自己的代码:

如果你分配了很多，你可以期待长时间的 GC 暂停，使用更小的分区可能会更好。

如果某段代码很昂贵，那么您的 shuffle 成本可以通过更高的并发性来摊销。

如果你有一个过滤器，你可以根据谓词的辨别力来调整分区的数量(如果你希望保留 5% 的数据和 99% 的数据，你会做出不同的决定)。

在我看来:

对于一次性作业，保留更多数量的分区以保持安全(慢总比失败好)。

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

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

通常，确定集群表现出稳定行为的每个分区的原始数据量相对容易(根据我的经验，它在几百兆字节的范围内，具体取决于用于加载数据的格式、数据结构、和配置)。这是您正在寻找的“魔数(Magic Number)”。

一般来说，您必须记住一些事情:

分区数不一定反射(reflect)
数据分布。任何需要随机播放的操作(*byKey、join、RDD.partitionBy、Dataset.repartition)都可能导致数据分布不均匀。始终监控您的工作是否存在严重数据倾斜的症状。

分区的数量一般不是恒定的。任何具有多个依赖项( union 、 coGroup 、 join )的操作都会影响分区数。