尝试覆盖 Hive 分区时写入 __HIVE_DEFAULT_PARTITION__ 的行已损坏

当尝试使用 Spark 2.3 覆盖 Hive 表中的分区时，我看到一些非常奇怪的行为

首先，我在构建 SparkSession 时设置以下设置：

.config("spark.sql.sources.partitionOverwriteMode", "dynamic")

然后，我将一些数据复制到新表中，并按 date_id 列进行分区。

ds
  .write
  .format("parquet")
  .option("compression", "snappy")
  .option("auto.purge", "true")
  .mode(saveMode)
  .partitionBy("date_id")
  .saveAsTable("tbl_copy")

我可以在HDFS中看到相关的date_id目录已经创建。

然后，我创建一个 DataSet，其中包含我希望覆盖的分区的数据，其中包含单个 date_id 的数据，并将其插入到 Hive 中，如下所示：

  ds
    .write
    .mode(SaveMode.Overwrite)
    .insertInto("tbl_copy")

作为完整性检查，我将相同的数据集写入新表。

      ds
        .write
        .format("parquet")
        .option("compression", "snappy")
        .option("auto.purge", "true")
        .mode(SaveMode.Overwrite)
        .saveAsTable("tmp_tbl")

tmp_tbl 中的数据与预期完全一致。

然而，当我查看 tbl_copy 时，我看到一个新的 HDFS 目录 `date_id=HIVE_DEFAULT_PARTITION

查询 tbl_cpy

SELECT * from tbl_copy WHERE date_id IS NULL

我看到应该插入分区 date_id=20180523 的行，但是 date_id 列为空，并且不相关的 row_changed 列已填充值 20180523。

看来插入 Hive 不知何故导致我的数据被破坏。将相同的数据集写入新表不会导致任何问题。

有人能解释一下吗？

因此看来分区列必须是数据集中的最后一个。

我通过将以下方法添加到 Dataset[T] 上解决了这个问题。

def partitionsTail(partitionColumns: Seq[String]) = {
  val columns = dataset.schema.collect{ case s if !partitionColumns.contains(s.name) => s.name} ++ partitionColumns

  dataset.select(columns.head, columns.tail: _*).as[T]
}