Spark数据帧写入方法编写了许多小文件

Question

我有一个相当简单的工作将日志文件转换为镶木地板.它正在处理1.1TB的数据(分为64MB - 128MB文件 - 我们的块大小为128MB),大约有12000个文件.

工作如下:

 val events = spark.sparkContext
  .textFile(s"$stream/$sourcetype")
  .map(_.split(" \\|\\| ").toList)
  .collect{case List(date, y, "Event") => MyEvent(date, y, "Event")}
  .toDF()

df.write.mode(SaveMode.Append).partitionBy("date").parquet(s"$path")

它使用通用模式收集事件,转换为DataFrame,然后写出镶木地板.

我遇到的问题是,这会在HDFS集群上造成一些IO爆炸,因为它试图创建如此多的小文件.

理想情况下,我想在分区'date'中只创建一些镶木地板文件.

控制它的最佳方法是什么？是通过使用'coalesce()'吗？

这将如何影响给定分区中创建的文件数量？它取决于我在Spark中使用了多少执行程序？(目前设定为100).

Answer 1

你必须重新分配你DataFrame的匹配分区DataFrameWriter

试试这个:

df
.repartition($"date")
.write.mode(SaveMode.Append)
.partitionBy("date")
.parquet(s"$path")

Answer 2

在 Python 中，您可以将Raphael 的 Roth 答案重写为：

(df
  .repartition("date")
  .write.mode("append")
  .partitionBy("date")
  .parquet("{path}".format(path=path)))

您还可以考虑添加更多列.repartition以避免非常大的分区出现问题：

(df
  .repartition("date", another_column, yet_another_colum)
  .write.mode("append")
  .partitionBy("date)
  .parquet("{path}".format(path=path)))