使用spark覆盖hive分区

Question

我正在使用AWS,我有使用Spark和Hive的工作流程.我的数据按日期分区,所以每天我在S3存储中都有一个新分区.我的问题是有一天加载数据失败并且我必须重新执行该分区.写的代码是下一个:

df                            // My data in a Dataframe
  .write
  .format(getFormat(target))  // csv by default, but could be parquet, ORC...
  .mode(getSaveMode("overwrite"))  // Append by default, but in future it should be Overwrite
  .partitionBy(partitionName) // Column of the partition, the date
  .options(target.options)    // header, separator...
  .option("path", target.path) // the path where it will be storage
  .saveAsTable(target.tableName)  // the table name

我的流程会发生什么？如果我使用SaveMode.Overwrite,整个表将被删除,我将只保存分区.如果我使用SaveMode.Append我可能有重复的数据.

进行搜索,我发现Hive支持这种覆盖,只有分区,但是使用hql语句,我没有它.

我们需要Hive上的解决方案,所以我们不能使用这个替代选项(直接到csv).

我发现这张Jira票据可以解决我遇到的问题,但尝试使用最新版本的Spark(2.3.0),情况是一样的.它删除整个表并保存分区,而不是覆盖我的数据所具有的分区.

为了使这更清楚,这是一个例子:

由A分区

数据:

| A | B | C | 
|---|---|---| 
| b | 1 | 2 | 
| c | 1 | 2 |

表:

| A | B | C | 
|---|---|---| 
| a | 1 | 2 | 
| b | 5 | 2 | 

我想要的是:在表中,分区a保留在表中,分区b覆盖数据,并添加分区c.有没有使用Spark的解决方案,我能做到这一点？

我这样做的最后一个选择是首先删除要保存的分区,然后使用SaveMode.Append,但如果没有其他解决方案,我会尝试这个.

Answer 1

如果您使用的是Spark 2.3.0,请尝试将设置spark.sql.sources.partitionOverwriteMode设置为dynamic,需要对数据集进行分区,并覆盖写入模式.

spark.conf.set("spark.sql.sources.partitionOverwriteMode","dynamic")
data.write.mode("overwrite").insertInto("partitioned_table")