spark.csv 如何确定读取的分区数？答案

【问题标题】：How does spark.csv determine the number of partitions on read?spark.csv 如何确定读取的分区数？
【发布时间】：2018-11-02 22:46:16
【问题描述】：

在 Spark 2.2.0 中：我正在使用

读取一个文件

spark.csv.read("filepath").load().rdd.getNumPartitions

我在一个系统中为 350 MB 文件获得 77 个分区，在另一个系统中获得 88 个分区。我还为 28 GB 文件获得了 226 个分区，大约为 28*1024 MB/128 MB 。问题是，Spark CSV 数据源是如何确定这个默认分区数的？

【问题讨论】：

请包含 Spark 版本。看起来您没有使用spark-csv，而是使用了内置的 Spark 阅读器。 This might be helpful 但不会完全回答您的问题。
根据建议更新问题！

标签： apache-spark

【解决方案1】：

分区数受多种因素影响——通常

spark.default.parallelism
正在读取的文件数（如果从目录读取文件）
影响spark.default.parallelism的集群管理器/核心数（参见spark configuration）

从文本文件（以及 CSV）读取时的分区数应根据CSVDataSource 确定为math.min(defaultParallelism, 2)

【讨论】：

这就是我一开始的想法，但是 spark.default.parallelism 设置为 100，从文档中，它主要适用于像 join 这样的转换。对于 'math.min(defaultParallelism, 2)` 来说是一件好事，但这并不能解释我是如何得到大于 2 的数字的。必须有别的东西来确定分区的数量。
了解更多关于您正在运行这些作业的系统的详细信息可能会很有趣，例如您使用的是哪种模式以及您是否只有一个文件或多个文件。

【解决方案2】：

从任何文件读取时的分区数遵循以下公式。

步骤1：从我在本地测试的指定路径查找文件大小/文件夹大小。您可以根据您的要求（s3/hdfs）找到。

import os
def find_folder_size(path):
    total = 0
    for entry in os.scandir(path):
        if entry.is_file():
            total += entry.stat().st_size
        elif entry.is_dir():
            total += find_folder_size(entry.path)
    return total

第二步：应用公式

target_partition_size = 200  #100 or 200 depends on your target partition
total_size = find_folder_size(paths)
print('Total size: {}'.format(total_size))
print(int(math.ceil(total_size / 1024.0 / 1024.0 / float(target_partition_size))))
num_partitions = int(math.ceil(total_size / 1024.0 / 1024.0 / float(target_partition_size)))
PARTITION_COLUMN_NAME = ['a','c']
df = df.repartition(num_partitions, PARTITION_COLUMN_NAME)
or 
df = df.repartition(num_partitions)

我们可以申请大数据/小数据来获取分区数。

【讨论】：

这个问题并不像我希望的那样清楚，但它并不是指通过 df.repartition 达到目标分区数。相反，它是关于 spark 如何确定默认分区数的机制。可以想象它会基于文件大小，但我们看到的行为是 2.2.0 中的 Spark 提供了不同数量的默认分区。值得注意的是，即使对于小文件大小，它也会提供大量分区，并且在具有相同大小的文件之间似乎有所不同（可能是在不同机器上的两个不同 spark 实例之间）。

【解决方案3】：

在读取 csv 文件（单个大文件或多个小文件，压缩与否）时，我发现spark.sql.files.maxPartitionBytes 对生成的分区数量有很大影响。调整这个值（默认为 128MB，参见 https://spark.apache.org/docs/latest/sql-performance-tuning.html）对我来说很关键。

【讨论】：