Spark parquet分区：大量文件

Question

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

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

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

为了避免我尝试过

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

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

我应该如何使用分区来避免写入后的许多文件？

Answer 1

首先我真的会避免使用coalesce，因为这通常会在转换链中被推到更远的位置，并且可能会破坏你工作的并行性（我在这里问过这个问题：Coalesce reduces parallelism of entire stage (spark)）

为每个 parquet-partition 写入 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 可以是数据集的另一个属性，或者是对现有属性使用一些模或舍入运算的派生属性。您甚至可以在 key 上使用带有 row_number 的窗口函数，然后将其四舍五入

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

这会将您的 N 记录放入 1 个镶木地板文件中

使用 orderBy

您还可以通过相应地对数据框进行排序来控制文件的数量而无需重新分区：

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

这将导致所有分区（默认为 200 个）总共（至少，但不多于）spark.sql.shuffle.partitions 个文件。在$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

Answer 2

这对我很有效：

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

它在每个输出分区（目录）中生成 N 个文件，并且（传闻）比使用 coalesce 更快 和（再次传闻，在我的数据集上）比仅在输出。

如果您正在使用 S3，我还建议您在本地驱动器上执行所有操作（Spark 在写出期间执行大量文件创建/重命名/删除），一旦完成，请使用 hadoop FileUtil（或仅使用 aws cli) 复制所有内容：

import java.net.URI
import org.apache.hadoop.fs.{FileSystem, FileUtil, Path}
// ...
  def copy(
          in : String,
          out : String,
          sparkSession: SparkSession
          ) = {
    FileUtil.copy(
      FileSystem.get(new URI(in), sparkSession.sparkContext.hadoopConfiguration),
      new Path(in),
      FileSystem.get(new URI(out), sparkSession.sparkContext.hadoopConfiguration),
      new Path(out),
      false,
      sparkSession.sparkContext.hadoopConfiguration
    )
  }

编辑：根据 cmets 中的讨论：

您的数据集的分区列为 YEAR，但每个给定的 YEAR 中的数据量都大不相同。因此，一年可能有 1GB 的数据，但另一年可能有 100GB。

这是处理此问题的一种方法的伪代码：

val partitionSize = 10000 // Number of rows you want per output file.
val yearValues = df.select("YEAR").distinct
distinctGroupByValues.each((yearVal) -> {
  val subDf = df.filter(s"YEAR = $yearVal")
  val numPartitionsToUse = subDf.count / partitionSize
  subDf.repartition(numPartitionsToUse).write(outputPath + "/year=$yearVal")
})

但是，我实际上不知道这会起作用。 Spark 可能会在每个列分区读取可变数量的文件时遇到问题。

另一种方法是编写您自己的自定义分区程序，但我不知道其中涉及什么，所以我无法提供任何代码。

Answer 3

让我们用另一种方法扩展 Raphael Roth 的答案，该方法将创建每个分区可以包含的文件数量的上限，as discussed in this answer：

import org.apache.spark.sql.functions.rand

df.repartition(numPartitions, $"some_col", rand)
  .write.partitionBy("some_col")
  .parquet("partitioned_lake")

Answer 4

这里的其他答案都很好但是有一些问题：

• 依靠maxRecordsPerFile 将大分区分解为较小的文件非常方便，但有两个注意事项：

  1. 如果您的分区列严重倾斜，则通过它们重新分区意味着可能会将最大数据分区的所有数据移动到单个 DataFrame 分区中。如果该 DataFrame 分区变得太大，仅此一项可能会使您的工作崩溃。

     举个简单的例子，想象一下repartition("country") 会为世界上每个人都有 1 行的 DataFrame 做什么。

  2. maxRecordsPerFile 将确保您的输出文件不超过一定的行数，但只有一个任务能够连续写出这些文件。一项任务必须处理整个数据分区，而不是能够写出包含多个任务的大型数据分区。

• repartition(numPartitions, $"some_col", rand) 是一个优雅的解决方案，但不能很好地处理小数据分区。它会为每个数据分区写出numPartitions 文件，即使它们很小。

  在许多情况下这可能不是问题，但如果您有一个大型数据湖，您就会知道写出许多小文件会随着时间的推移降低数据湖的性能。

因此，一种解决方案不适用于非常大的数据分区，而另一种解决方案不适用于非常小的数据分区。

我们需要一种根据数据分区大小动态扩展输出文件数量的方法。如果它非常大，我们需要很多文件。如果它非常小，我们只需要几个文件，甚至只需要一个文件。

解决方案是使用repartition(..., rand) 扩展该方法，并根据该数据分区所需的输出文件数动态扩展rand 的范围。

这是the solution I posted 在一个非常相似的问题上的精髓：

# In this example, `id` is a column in `skewed_data`.
partition_by_columns = ['id']
desired_rows_per_output_file = 10

partition_count = skewed_data.groupBy(partition_by_columns).count()

partition_balanced_data = (
    skewed_data
    .join(partition_count, on=partition_by_columns)
    .withColumn(
        'repartition_seed',
        (
            rand() * partition_count['count'] / desired_rows_per_output_file
        ).cast('int')
    )
    .repartition(*partition_by_columns, 'repartition_seed')
)

这将平衡输出文件的大小，而不考虑分区倾斜，并且不会限制您的并行度或为小分区生成太多小文件。

如果您想自己运行此代码，我提供了 a self-contained example，以及 DataFrame 分区正确平衡的证明。