如何将Spark DF作为新列追加到现有Hive表中

核心结论 #

• 不能直接将带新列的Spark DF写入现有Hive表，默认状态下必须先执行ALTER TABLE ADD COLUMN语句更新Hive表的Schema后再写入。
• 如果你使用的是Spark 2.3+，且Hive表的存储格式是Parquet、ORC这类支持Schema演进的格式，可以通过开启配置实现自动新增列，无需手动执行ALTER语句。

方案1：手动更新Schema（兼容性最好，推荐） #

这是最稳妥的方案，不受存储格式、Spark版本限制，步骤如下：

1. 比对Spark DF和现有Hive表的列差异，找出新增的列
2. 执行ALTER TABLE语句添加新列，注意列类型要和DF中对应列的类型一致
3. 再执行原写入逻辑即可

示例代码：

from pyspark.sql.functions import lit

hive_table_fullname = f"{datalab}.{table}"
# 读取现有Hive表的列
hive_cols = set(spark.table(hive_table_fullname).columns)
df_cols = set(temp.columns)
# 筛选出新增列
new_cols = [col for col in temp.columns if col not in hive_cols]

if new_cols:
    # 拼接ADD COLUMN语句，从DF schema中自动读取列类型
    for col_name in new_cols:
        col_type = temp.schema[col_name].dataType.simpleString()
        spark.sql(f"ALTER TABLE {hive_table_fullname} ADD COLUMNS ({col_name} {col_type})")

# 执行写入
temp.write.mode('overwrite').insertInto(hive_table_fullname, overwrite=True)

方案2：开启自动Schema演进配置（仅支持特定存储格式） #

如果你的Hive表使用Parquet/ORC格式，且Spark版本在2.3及以上，可以开启如下配置，写入时会自动将DF中的新列同步到Hive表Schema中：

# 开启Parquet Schema自动合并
spark.conf.set("spark.sql.parquet.mergeSchema", "true")
# 允许写入时自动新增列（Spark 3.0+支持）
spark.conf.set("spark.sql.legacy.allowAddingNewColumnsDuringInsert", "true")
# Hive catalog适配配置
spark.conf.set("spark.sql.hive.caseSensitiveInferenceMode", "INFER_AND_SAVE")

配置开启后直接执行原写入语句即可。

注意事项 #

• insertInto是按列位置匹配写入，不是按列名匹配，如果DF列顺序和Hive表列顺序不一致会导致数据错位，推荐改用saveAsTable配合Schema合并参数，按列名匹配更稳妥：

temp.write.mode("overwrite").option("mergeSchema", "true").saveAsTable(hive_table_fullname)

• 上述方法仅支持新增非分区列，如果是分区表需要调整分区列，需要全量重写表完成更新。

内容的提问来源于stack exchange，提问作者Víctor Manuel Nájera Vargas