将多个小文件合并到Spark中的几个较大文件中

Question

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

'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？

Answer 1

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

Answer 2

我遇到过同样的问题。解决方案是使用分区列添加DISTRIBUTE BY子句。这可确保一个分区的数据转到单个reducer。你案例中的例子：

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

Answer 3

数据框repartition(1)方法适用于这种情况。