Spark数据帧写入方法写入许多小文件答案

【问题标题】：Spark dataframe write method writing many small filesSpark数据帧写入方法写入许多小文件
【发布时间】：2017-11-11 13:31:58
【问题描述】：

我有一个相当简单的工作，将日志文件转换为镶木地板。它正在处理 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，然后以 parquet 形式写出。

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

理想情况下，我只想在分区“日期”内创建少量镶木地板文件。

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

这将如何影响在给定分区中创建的文件数量？它是否取决于我在 Spark 中工作的执行者数量？（目前设置为 100）。

【问题讨论】：

与问题无关，但你不应该收集你的数据（第一条语句），而是在你的RDD上使用map
@RaphaelRoth 这个收藏是不同的。这更像是过滤器 -> 映射github.com/apache/spark/blob/v2.1.1/core/src/main/scala/org/…
@user3030878 您是如何让Spark 准确写入 64 MB / 128 MB 文件的？我的Spark 工作提供了微小的（每个 1-2 MB）文件（文件数 = 默认值 = 200）。我不能简单地调用 repartition(n) 来获得每个大约 128 MB 的文件，因为 n 在不同的工作中会有很大的不同。

标签： scala apache-spark

【解决方案1】：

您必须重新分区您的 DataFrame 以匹配 DataFrameWriter 的分区

试试这个：

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

【讨论】：

如何从多个根目录读取 parquet 文件并合并它们？
@Raphael partitionBy(date) 不会减少分区以计算不同日期吗？我认为DF有相同的编号。分区作为不同的日期计数。我说的对吗？
我过去也注意到 df .repartition(n,$"date") 也给出了相同的结果。这只是偶然吗？大体上查找 API 接口等不那么合乎逻辑。

【解决方案2】：

在 Python 中，您可以将 Raphael's Roth answer 重写为：

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

【讨论】：

【解决方案3】：

最简单的解决方案是将您的实际分区替换为：

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

您还可以为您的DataFrame 使用更精确的分区，即一天或者一个小时的小时范围。然后你可以对作家不那么精确。这实际上取决于数据量。

您可以通过对DataFrame 进行分区和使用 partition by 子句进行写入来减少熵。

【讨论】：

【解决方案4】：

我遇到了同样的问题，我可以使用 coalesce 解决我的问题。

df
  .coalesce(3) // number of parts/files 
  .write.mode(SaveMode.Append)
  .parquet(s"$path")

更多关于使用coalesce或repartition的信息可以参考以下spark: coalesce or repartition

【讨论】：

【解决方案5】：

从这里复制我的答案：https://stackoverflow.com/a/53620268/171916

这对我很有效：

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

【讨论】：

【解决方案6】：

尝试运行这样的脚本作为地图作业如何将所有拼花文件合并为一个：

$ hadoop jar /usr/hdp/2.3.2.0-2950/hadoop-mapreduce/hadoop-streaming-2.7.1.2.3.2.0-2950.jar \
 -Dmapred.reduce.tasks=1 \
 -input "/hdfs/input/dir" \
 -output "/hdfs/output/dir" \
 -mapper cat \
 -reducer cat

【讨论】：