在 Spark 中将多个小文件合并为几个大文件答案

【问题标题】：merge multiple small files in to few larger files in Spark在 Spark 中将多个小文件合并为几个大文件
【发布时间】：2015-09-09 16:02:41
【问题描述】：

我通过 Spark 使用 hive。我的 spark 代码中有一个 Insert into partitioned table 查询。输入数据为 200+gb。当 Spark 写入分区表时，它会吐出非常小的文件（以 kb 为单位的文件）。所以现在输出分区表文件夹有 5000+ 小 kb 文件。我想将这些合并到几个大 MB 文件中，可能是几个 200mb 的文件。我厌倦了使用 hive 合并设置，但它们似乎不起作用。

'val result7A = hiveContext.sql("set hive.exec.dynamic.partition=true")

 val result7B = hiveContext.sql("set hive.exec.dynamic.partition.mode=nonstrict")

val result7C = hiveContext.sql("SET hive.merge.size.per.task=256000000")

val result7D = hiveContext.sql("SET hive.merge.mapfiles=true")

val result7E = hiveContext.sql("SET hive.merge.mapredfiles=true")

val result7F = hiveContext.sql("SET hive.merge.sparkfiles = true")

val result7G = hiveContext.sql("set hive.aux.jars.path=c:\\Applications\\json-serde-1.1.9.3-SNAPSHOT-jar-with-dependencies.jar")

val result8 = hiveContext.sql("INSERT INTO TABLE partition_table PARTITION (date) select a,b,c from partition_json_table")'

上述配置单元设置在 mapreduce 配置单元执行中工作，并输出指定大小的文件。是否有任何选项可以执行此 Spark 或 Scala？

【问题讨论】：

所以您目前每天分区获得很多文件，而您只想要几个？还是要将多天合并到一个文件中？
@maxymoo 我每天分区都会收到很多小文件（kb 文件），而我只想要一些大文件（以 mbs 为单位）。
嗨@user3267086 你能解决这个小文件的问题吗？我尝试使用 hc.sql("bla bla").coalesce(10) 方法，但我仍然看到 200 个大约 20 MB 的小文件

标签： scala hadoop apache-spark hive apache-spark-sql

【解决方案1】：

我有同样的问题。解决方案是在分区列中添加 DISTRIBUTE BY 子句。这确保了一个分区的数据进入单个 reducer。你的例子：

INSERT INTO TABLE partition_table PARTITION (date) select a,b,c from partition_json_table DISTRIBUTE BY date

【讨论】：

嘿@Jussi Kujala 非常感谢，这似乎对我有用。但是我有一个问题，如果我的表被不止一列分区怎么办......在这种情况下也可以吗？

【解决方案2】：

您可能想尝试使用DataFrame.coalesce 方法；它返回一个具有指定数量的分区的 DataFrame（每个分区在插入时成为一个文件）。因此，使用您要插入的记录数和每条记录的典型大小，如果您想要约 200MB 的文件，您可以估计要合并到多少个分区。

【讨论】：

嗨@zweiterlinde 我尝试使用 hc.sql("bla bla").coalesce(10) 方法，但我仍然看到 200 个大约 20 MB 的小文件
我需要更长的代码示例才能真正能够发表评论，但在我的玩具实验中，调用 df.write.parquetFile(...) 会产生许多部分文件，但 df.coalesce(1).write.parquetFile(...) 会产生一个。

【解决方案3】：

数据框repartition(1) 方法在这种情况下有效。

【讨论】：