Spark 2.2.0实现特定分区覆盖的替代配置或方案求助

Spark 2.2.0不支持spark.sql.sources.partitionOverwriteMode=dynamic配置，直接使用overwrite模式会全量覆盖所有分区，以下是几种可行的替代方案：

方案1：手动删除目标分区后追加写入 #

先定位要覆盖的分区目录并删除，再以append模式写入数据，仅更新目标分区，其余分区保留。

代码示例（Python）： #

from pyspark.sql import SparkSession
from org.apache.hadoop.fs import Path

# 初始化Spark会话
spark = SparkSession.builder.appName("PartitionOverwrite").getOrCreate()
sc = spark.sparkContext

# 假设df是待写入的目标数据
df = spark.read.table("your_source_table").filter("partition_date='2024-05-20' AND hour='12'")

# 获取需要覆盖的分区列表
target_partitions = df.select("partition_date", "hour").distinct().collect()

# 获取HDFS文件系统实例
fs = Path.get(sc._jsc.hadoopConfiguration())
base_output_path = "/user/test/test/output/"

# 删除目标分区目录
for partition in target_partitions:
    partition_path = f"{base_output_path}/partition_date={partition.partition_date}/hour={partition.hour}"
    hdfs_path = Path(partition_path)
    if fs.exists(hdfs_path):
        fs.delete(hdfs_path, recursive=True)

# 以追加模式写入数据
df.write \
  .mode("append") \
  .format("csv") \
  .partitionBy("partition_date", "hour") \
  .save(base_output_path)

方案2：使用Hive INSERT OVERWRITE语句（适用于Hive表场景） #

如果数据写入Hive表（管理表或外部表），可利用Hive的INSERT OVERWRITE语法指定分区，仅覆盖目标分区数据，这在Spark 2.2.0中完全支持。

代码示例（Python）： #

# 将DataFrame注册为临时视图
df.createOrReplaceTempView("temp_data")

# 获取目标分区列表
target_partitions = df.select("partition_date", "hour").distinct().collect()

# 循环执行分区覆盖
for partition in target_partitions:
    partition_date = partition.partition_date
    hour = partition.hour
    spark.sql(f"""
        INSERT OVERWRITE TABLE your_hive_table
        PARTITION (partition_date='{partition_date}', hour='{hour}')
        SELECT col1, col2, partition_date, hour 
        FROM temp_data
        WHERE partition_date='{partition_date}' AND hour='{hour}'
    """)

注意事项： #

• 确保Hive表已正确定义partition_date、hour为分区列；
• 若为外部表，表的存储路径需与目标输出路径一致。

方案3：批量动态分区覆盖（开启Hive动态分区） #

如果需要批量覆盖多个分区，可开启Hive动态分区，结合INSERT OVERWRITE自动覆盖DataFrame中包含的所有分区，其余分区不受影响。

配置及代码示例： #

# 开启Hive动态分区并设置为非严格模式
spark.conf.set("hive.exec.dynamic.partition", "true")
spark.conf.set("hive.exec.dynamic.partition.mode", "nonstrict")

# 将DataFrame注册为临时视图
df.createOrReplaceTempView("temp_data")

# 执行动态分区覆盖
spark.sql(f"""
    INSERT OVERWRITE TABLE your_hive_table
    PARTITION (partition_date, hour)
    SELECT col1, col2, partition_date, hour 
    FROM temp_data
""")

内容的提问来源于stack exchange，提问作者Rahul Patidar