scala - 使用 spark 覆盖配置单元分区

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 上的解决方案，所以我们不能使用这个 alternative option (直接到 csv)。

我找到了这个 Jira ticket假设可以解决我遇到的问题，但是尝试使用最新版本的 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")