hadoop - 如何在写入 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() 方法将数据写入文件，不接触元信息。所以它不能覆盖 HIVE 中的表。

如果您想覆盖 hive 表中的数据，请使用方法 .insertInto(hive_table_name)，其中 hive_table_name 是 HIVE 中表的全名(schema + table_name)

以你为例

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

也可以用元数据信息覆盖数据。带有覆盖修饰符的方法 .saveAsTable(hive_table_name) 也会覆盖 Metastore 中的数据。

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