pyspark 在输出中写入大量较小的文件

Question

我正在使用 pyspark 处理一些数据并将输出写入 S3。我在 athena 中创建了一个表，用于查询这些数据。

数据是json字符串的形式（每行一个），spark代码读取文件，根据某些字段对其进行分区并写入S3。

对于 1.1 GB 的文件，我看到 spark 正在写入 36 个文件，每个文件大小约为 5 MB。在阅读 athena 文档时，我发现最佳文件大小为 ~128 MB。 https://aws.amazon.com/blogs/big-data/top-10-performance-tuning-tips-for-amazon-athena/

sparkSess = SparkSession.builder\
    .appName("testApp")\
    .config("spark.debug.maxToStringFields", "1000")\
    .config("spark.sql.sources.partitionOverwriteMode", "dynamic")\
    .getOrCreate()

sparkCtx = sparkSess.sparkContext
deltaRdd = sparkCtx.textFile(filePath)
df = sparkSess.createDataFrame(deltaRdd, schema)

try:
    df.write.partitionBy('field1','field2','field3')\
        .json(path, mode='overwrite', compression=compression)
except Exception as e:
    print (e)

为什么 spark 会写这么小的文件。有什么方法可以控制文件大小。

Answer 1

有没有办法控制文件大小？

有一些控制机制。但是，它们并不明确。

s3 驱动程序不是 spark 本身的一部分。它们是 spark emr 附带的 hadoop 安装的一部分。 s3 块大小可以设置在 /etc/hadoop/core-site.xmlconfig file。

但默认情况下它应该是 128 mb 左右。

为什么 spark 会写这么小的文件

Spark 将遵守 hadoop 块大小。但是，您可以在写作之前使用partionBy。

假设您使用partionBy("date").write.csv("s3://products/")。 Spark 将为每个分区创建一个带有date 的子文件夹。之内 每个分区文件夹 spark 将再次尝试创建块并尝试遵守fs.s3a.block.size。

例如

s3:/products/date=20191127/00000.csv
s3:/products/date=20191127/00001.csv
s3:/products/date=20200101/00000.csv

在上面的示例中 - 特定分区可以小于 128mb 的块大小。

所以只需仔细检查/etc/hadoop/core-site.xml 中的块大小，以及在写入之前是否需要使用partitionBy 对数据帧进行分区。

编辑：

Similar post 还建议重新分区数据帧以匹配partitionBy 方案

df.repartition('field1','field2','field3')
.write.partitionBy('field1','field2','field3')

writer.partitionBy 对现有数据帧分区进行操作。它不会 repartition 原始数据框。因此，如果整个数据帧的分区不同，就会发生嵌套分区。