Spark Dataframe 在覆盖 Hive 表的分区数据时出现问题

Question

下面是我的 Hive 表定义：

CREATE EXTERNAL TABLE IF NOT EXISTS default.test2(
id integer,
count integer
)
PARTITIONED BY (
fac STRING,
fiscaldate_str DATE )
STORED AS PARQUET
LOCATION 's3://<bucket name>/backup/test2';

我有如下配置单元表中的数据，（我刚刚插入了示例数据）

select * from default.test2

+---+-----+----+--------------+
| id|count| fac|fiscaldate_str|
+---+-----+----+--------------+
|  2|    3| NRM|    2019-01-01|
|  1|    2| NRM|    2019-01-01|
|  2|    3| NRM|    2019-01-02|
|  1|    2| NRM|    2019-01-02|
|  2|    3| NRM|    2019-01-03|
|  1|    2| NRM|    2019-01-03|
|  2|    3|STST|    2019-01-01|
|  1|    2|STST|    2019-01-01|
|  2|    3|STST|    2019-01-02|
|  1|    2|STST|    2019-01-02|
|  2|    3|STST|    2019-01-03|
|  1|    2|STST|    2019-01-03|
+---+-----+----+--------------+

该表在两列（fac、Fiscaldate_str）上进行分区，我们正尝试使用 spark 数据帧 - 数据帧编写器在分区级别动态执行插入覆盖。

但是，在尝试此操作时，我们要么以重复数据结束，要么所有其他分区都被删除。

以下是使用 spark 数据框的代码片段。

首先，我正在创建数据框

df = spark.createDataFrame([(99,99,'NRM','2019-01-01'),(999,999,'NRM','2019-01-01')], ['id','count','fac','fiscaldate_str'])

df.show(2,False)
+---+-----+---+--------------+
|id |count|fac|fiscaldate_str|
+---+-----+---+--------------+
|99 |99   |NRM|2019-01-01    |
|999|999  |NRM|2019-01-01    |
+---+-----+---+--------------+

1. 使用以下代码段重复，

   df.coalesce(1).write.mode("overwrite").insertInto("default.test2")

2. 所有其他数据都被删除，只有新数据可用。

   df.coalesce(1).write.mode("overwrite").saveAsTable("default.test2")

或者

   df.createOrReplaceTempView("tempview")

tbl_ald_kpiv_hist_insert = spark.sql("""
INSERT OVERWRITE TABLE default.test2 
partition(fac,fiscaldate_str) 
select * from tempview
""")

我将 AWS EMR 与 Spark 2.4.0 和 Hive 2.3.4-amzn-1 以及 S3 一起使用。

谁能知道为什么我不能将数据动态覆盖到分区中？

Answer 1

你的问题不太容易理解，但我认为你的意思是你想要覆盖一个分区。如果是这样，那么这就是你所需要的，你所需要的——第二行：

df = spark.createDataFrame([(99,99,'AAA','2019-01-02'),(999,999,'BBB','2019-01-01')], ['id','count','fac','fiscaldate_str'])
df.coalesce(1).write.mode("overwrite").insertInto("test2",overwrite=True) 

注意覆盖=真。所做的评论既不在这里也不在那里，因为正在使用 DF.writer。我不是在解决合并（1）。

对提问者的评论

我按照我的标准运行它 - 在进行原型设计和回答时 - 在 Databricks Notebook 上并明确设置以下内容并且它工作正常：

spark.conf.set("spark.sql.sources.partitionOverwriteMode","static")
spark.conf.set("hive.exec.dynamic.partition.mode", "strict")

您要求使用以下内容更新答案：

spark.conf.set("spark.sql.sources.partitionOverwriteMode","d??ynamic").

可以像我刚做的那样做；可能在你的环境中这是需要的，但我确实不需要这样做。

更新 19/3/20

这适用于先前的 Spark 版本，现在适用于以下版本：

spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")
// In Databricks did not matter the below settings
//spark.conf.set("hive.exec.dynamic.partition", "true")
//spark.conf.set("hive.exec.dynamic.partition.mode", "nonstrict")

Seq(("CompanyA1", "A"), ("CompanyA2", "A"), 
    ("CompanyB1", "B"))
.toDF("company", "id")
.write
.mode(SaveMode.Overwrite)
.partitionBy("id")
.saveAsTable("KQCAMS9")

spark.sql(s"SELECT * FROM KQCAMS9").show(false)

val df = Seq(("CompanyA3", "A"))
.toDF("company", "id")
// disregard coalsece
df.coalesce(1).write.mode("overwrite").insertInto("KQCAMS9") 

spark.sql(s"SELECT * FROM KQCAMS9").show(false)
spark.sql(s"show partitions KQCAMS9").show(false)

现在从 2.4.x 开始就这样了。向前。