Spark拼花地板分区:大量文件

首先,我真的会避免使用 coalesce ,因为这通常在转换链中被进一步推高,可能会破坏您工作的并行性(我在这里询问了这个问题: Coalesce reduces parallelism of entire stage (spark)

每个拼花地板分区写一个文件非常容易(请参见 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

使用orderBy

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

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

这将导致总计(至少,但不超过) spark.sql.shuffle.partitions $key ,因为parquet将记住数据帧的顺序,并相应地写入统计信息。例如,您可以按ID订购:

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

这不会更改文件的数量,但在查询给定的拼花地板文件时会提高性能 钥匙 和 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

火花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

让我们用一种额外的方法来扩展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")

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

它在每个输出分区(目录)中生成N个文件,并且(据说)比使用 coalesce 和 (有趣的是,在我的数据集上)比只对输出进行重新分区更快。

如果你在使用S3,我还建议你在本地驱动器上做所有事情(Spark在写操作中会创建/重命名/删除很多文件),一旦一切都解决了,就使用hadoop FileUtil

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
    )
  }

编辑:根据评论中的讨论:

您可以创建一个分区列为年份的数据集,但每个给定年份中的数据量相差很大。因此,一年可能有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可能会在读取每个列分区中数量可变的文件时遇到问题。

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

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

  1. 如果分区列严重倾斜,则通过它们进行重新分区意味着可能将最大数据分区的所有数据移动到单个数据帧分区中。如果数据帧分区太大,那么单是这一点就可能使您的工作崩溃。

     repartition("country") 对于世界上每个人都有一行的数据帧就可以了。

  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 ,以及数据帧分区正确平衡的证明。