如何在现有作业中从Stage表向Delta最终表添加带值新列？

先理清楚你的场景：Stage表已经接到了源端带新增字段的数据，现在要把它合并到Delta最终表里，同时自动兼容这些新增字段——核心就是开启并使用Delta的Merge Schema功能，下面是一步步的实操：

1. 开启最终Delta表的Schema Merge功能 #

Delta默认关闭Schema Merge，得先给最终表开这个开关，两种方式：

方式一：给已存在的表修改配置 #

ALTER TABLE delta.`最终表的存储路径` SET TBLPROPERTIES ('delta.enableSchemaMerge' = 'true')

方式二：创建表时直接配置 #

CREATE TABLE 最终表名 (字段1 string, 字段2 int)
USING DELTA
TBLPROPERTIES ('delta.enableSchemaMerge' = 'true')

2. 写Merge语句完成数据+Schema合并 #

用Delta标准的Merge语法，只要开了Schema Merge，Stage表的新增字段会自动被合并到最终表里，同步数据：

MERGE INTO 最终表名 t
USING Stage表名 s
ON t.主键字段 = s.主键字段  -- 替换成你的实际关联条件，比如订单ID、用户ID这类唯一匹配键
WHEN MATCHED THEN
  UPDATE SET *  -- 全量更新匹配行，自动包含新增字段
WHEN NOT MATCHED THEN
  INSERT *  -- 插入未匹配的全量数据，自动带上新增字段

要是不想全量更新/插入，也可以指定具体字段，但必须把新增字段写进去，用*是最省心的方式，自动适配所有字段变化。

3. 作业化运行的关键配置 #

如果是用Spark作业跑这个逻辑，要确保SparkSession加载了Delta的扩展：

// Scala 示例代码
val spark = SparkSession.builder()
  .appName("MergeStageToDeltaFinal")
  .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
  .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
  .getOrCreate()

调度作业的时候不用额外判断有没有新增字段——只要Stage表有新字段，Merge操作会自动识别并合并，不用手动改最终表结构。

4. 验证合并结果 #

跑完作业后，查一下最终表的结构和数据，确认新增字段已经加上且数据正确：

-- 查看表结构
DESCRIBE TABLE 最终表名;
-- 验证新增字段的数据
SELECT 新增字段名 FROM 最终表名 WHERE 主键字段 = '测试用的匹配值';

内容的提问来源于stack exchange，提问作者Saswat Ray