如何在将小文件写入hive orc表的同时合并spark中的小文件

Question

我正在从s3读取csv文件，并以ORC的身份写入配置单元表。在写的同时，它也在写大量的小文件。我需要合并所有这些文件。我设置了以下属性:

 spark.sql("SET hive.merge.sparkfiles = true")
 spark.sql("SET hive.merge.mapredfiles = true")
 spark.sql("SET hive.merge.mapfiles = true")
 spark.sql("set hive.merge.smallfiles.avgsize = 128000000")
 spark.sql("set hive.merge.size.per.task = 128000000")

除了这些配置之外，我尝试了repartition(1)和coalesce(1)，这将合并到单个文件中，但它会删除配置单元表并重新创建它。

 masterFile.repartition(1).write.mode(SaveMode.Overwrite).partitionBy(<partitioncolumn>).orc(<HIVEtablePath>);

如果我使用追加模式而不是覆盖模式，它会在每个分区下创建重复文件。

  masterFile.repartition(1).write.mode(SaveMode.Append).partitionBy(<partitioncolumn>).orc(<HIVEtablePath>);

在这两种情况下，spark作业运行两次，在第二次执行时失败。

有没有任何方法可以使用附加模式的重新分区/合并，而不重复每个分区中的部分文件？

Answer 1

masterFile.repartition(1).write.mode(SaveMode.Overwrite).partitionBy(<partitioncolumn>).orc(<HIVEtablePath>)

.orc（）方法将数据作为文件写入，不触及元信息。因此无法覆盖配置单元中的表。

如果要覆盖配置单元表中数据，请使用method.insertinto(hive_table_name)，其中hive_table_name是配置单元中表的全名（模式+表名）

根据你的例子

masterFile.repartition(1).write.mode(SaveMode.Overwrite).partitionBy(<partitioncolumn>).insertInto(hiveTableName)

还可以用元数据信息覆盖数据。具有覆盖修饰符的saveastable(hive_table_name)也将覆盖转移的数据。

masterFile.repartition(1).write.mode(SaveMode.Overwrite).partitionBy(<partitioncolumn>).saveAsTable(hiveTableName)