apache-spark - Spark Parquet 分区 : Large number of files

Question

我正在尝试利用 Spark 分区。我试图做类似的事情

data.write.partitionBy("key").parquet("/location")

这里的问题每个分区都会创建大量的 Parquet 文件，如果我试图从根目录读取，会导致读取速度变慢。

为了避免我试过

data.coalese(numPart).write.partitionBy("key").parquet("/location")

然而，这会在每个分区中创建 numPart 数量的 Parquet 文件。
现在我的分区大小不同了。所以我理想情况下希望每个分区都有单独的合并。然而，这看起来并不是一件容易的事情。我需要访问所有分区合并到某个数量并存储在一个单独的位置。

我应该如何使用分区来避免写入后出现很多文件？

Answer 1

首先我真的会避免使用 coalesce ，因为这通常在转换链中被进一步推高，并且可能会破坏您工作的并行性(我在这里询问了这个问题:Coalesce reduces parallelism of entire stage (spark))
每个 Parquet 分区写入 1 个文件非常容易(请参阅 Spark dataframe write method writing many small files ):

data.repartition($"key").write.partitionBy("key").parquet("/location")

如果您想设置任意数量的文件(或具有相同大小的文件)，您需要使用另一个可以使用的属性进一步重新分区您的数据(我无法告诉您这可能是什么情况):

data.repartition($"key",$"another_key").write.partitionBy("key").parquet("/location")

another_key可以是数据集的另一个属性，也可以是对现有属性使用一些模运算或舍入运算的派生属性。您甚至可以将窗口函数与 row_number 一起使用在 key然后用类似的东西把它四舍五入

data.repartition($"key",floor($"row_number"/N)*N).write.partitionBy("key").parquet("/location")

这会让你 N记录到 1 个 Parquet 文件中
使用 orderBy
您还可以通过相应地对数据框进行排序来控制文件数量而无需重新分区:

data.orderBy($"key").write.partitionBy("key").parquet("/location")

这将导致总共(至少，但不多于) spark.sql.shuffle.partitions跨所有分区的文件(默认为 200)。在 $key 之后添加第二个排序列甚至是有益的, 因为 parquet 会记住数据帧的顺序并相应地写入统计信息。例如，您可以按 ID 订购:

data.orderBy($"key",$"id").write.partitionBy("key").parquet("/location")

这不会改变文件的数量，但是当您查询给定的 Parquet 文件时，它会提高性能 key和 id .见例如 https://www.slideshare.net/RyanBlue3/parquet-performance-tuning-the-missing-guide和 https://db-blog.web.cern.ch/blog/luca-canali/2017-06-diving-spark-and-parquet-workloads-example
Spark 2.2+
从 Spark 2.2 开始，您还可以使用新选项 maxRecordsPerFile限制每个文件的记录数 如果文件太大 .如果您有 N 个分区，您仍然至少会得到 N 个文件，但是您可以将 1 个分区(任务)写入的文件拆分为更小的块:

df.write
.option("maxRecordsPerFile", 10000)
...

见例如 http://www.gatorsmile.io/anticipated-feature-in-spark-2-2-max-records-written-per-file/和 spark write to disk with N files less than N partitions