如何在 Glue 中控制 Parquet 文件的大小？答案

【问题标题】：How to control size of Parquet files in Glue?如何在 Glue 中控制 Parquet 文件的大小？
【发布时间】：2020-02-03 16:09:41
【问题描述】：

我正在将数据集加载到 DynamicFrame 中，执行转换，然后将其写回 S3：

datasink = glueContext.write_dynamic_frame.from_options(
    frame = someDateFrame, 
    connection_type = "s3", 
    connection_options = {
        "path": "s3://the-bucket/some-data-set"
    }, 
    format = "parquet"
)

结果是 12 个 Parquet 文件，平均大小约为 3MB。

首先，鉴于几乎所有消费软件（Presto/Athena、Spark）都喜欢大约 100MB 的文件大小并且不是一堆小文件。如果有人对此有见解，我将不胜感激。

但实际上我想知道是否可以让 Glue/Spark 生成一个大文件或至少更大的文件。这可能吗？

【问题讨论】：

您可以正确指定 spark.sql.shuffle.partitions 而不是使用默认的 200。请参阅stackoverflow.com/questions/50701265/…

标签： apache-spark pyspark parquet aws-glue

【解决方案1】：

我还没试过。但是你可以在write_from_options中设置accumulator_size。

查看https://github.com/awslabs/aws-glue-libs/blob/master/awsglue/context.py 了解如何传递值。

或者，您可以在写入前使用带有 1 个分区的 pyspark DF，以确保它只写入一个文件。

df.coalesce(1).write.format('parquet').save('s3://the-bucket/some-data-set')

请注意，写入 1 个文件不会利用并行写入，因此会增加写入时间。

【讨论】：

设置accumulator_size对输出文件大小没有影响。

【解决方案2】：

您可以在将动态数据帧写入 S3 之前尝试 repartition(1)。请参阅 here 以了解为什么 coalesce(1) 是合并的错误选择。如果单个节点无法容纳所有要写入的数据，也可能导致内存不足（OOM）异常。

【讨论】：

【解决方案3】：

从长远来看，使用 coalesce(1) 会降低 Glue 的性能。虽然它可能适用于小文件，但对于较大的文件则需要非常长的时间。

coalesce(1) 只使用 1 个 spark 执行器来写入文件，如果没有 coalesce()，它会使用所有 spark 执行器来写入文件。

另外，使用 coalesce(1) 的成本会更高。 1 个执行器长时间运行所花费的成本将比所有执行器运行时间仅为 1 个执行器所用时间的一小部分。

Coalesce(1) 花了 4 小时 48 分钟来处理 1GB 的 Parquet Snappy 压缩数据。

Coalesce(9) 同样用了 48 分钟。

没有 Coalesce() 在 25 分钟内完成相同的工作。

【讨论】：

【解决方案4】：

我不明白为什么 Glue/Spark 默认不会创建一个大约 36MB 大的文件，因为几乎所有消费软件（Presto/Athena、Spark）都喜欢大约 100MB 的文件大小而不是一堆小文件。

输出文件的数量与分区的数量直接相关。 Spark 不能假定输出文件的默认大小，因为它取决于应用程序。控制输出文件大小的唯一方法是对分区编号进行操作。

我想知道是否可以让 Glue/Spark 生成一个大文件或至少更大的文件。这可能吗？

是的，这是可能的，但没有经验法则。您必须根据您的数据尝试不同的设置。如果您使用 AWS Glue API [1]，您可以在读取数据时控制如何将小文件分组到单个分区中：

glueContext.write_dynamic_frame.from_options(
frame = someDateFrame, 
connection_type = "s3", 
connection_options = {
    "path": "s3://the-bucket/some-data-set",
    "groupFiles": "inPartition", 
    "groupSize": "10485760" # 10485760 bytes (10 MB)
}
format = "parquet"

)

如果您的转换代码不会对数据分布产生太大影响（不过滤、不连接等），您应该期望输出文件的大小与输入的读取文件大小几乎相同（不考虑压缩率）一般来说, Spark 转换非常复杂，包括连接、聚合、过滤。这会改变数据分布和最终分区的数量。

在这种情况下，您应该使用coalesce() 或repartition() 来控制您期望的分区数。

[1]https://aws.amazon.com/premiumsupport/knowledge-center/glue-job-output-large-files/

【讨论】：