如何使用Python为Delta Table添加列（保留历史与Schema演进）

我有一个通过以下代码创建的Delta表：

# Load the data from its source.
df = spark.read.load("/databricks-datasets/learning-spark-v2/people/people-10m.delta")

# Write the data to a table.
table_name = "people_10m"
df.write.saveAsTable(table_name)

现在需要动态给这个表添加单个列、多列甚至嵌套数组列，已经通过Python的set API识别出要新增的列，要求实现时不读取全量数据重写、不丢失表历史记录，可以用Schema演进的方式，同时要禁止列删除。

解决方案：用DeltaTable API或SQL直接修改Schema #

Delta Lake支持直接修改表Schema，无需重写数据，还能保留所有历史版本，完全符合需求。

方法1：使用DeltaTable Python API

直接通过DeltaTable对象执行ALTER操作，动态生成新增列的语句：

from delta.tables import DeltaTable

# 获取目标Delta表对象
delta_table = DeltaTable.forName(spark, "people_10m")

# 假设这是你通过set API识别出的新增列列表，包含普通列、嵌套结构列、数组列
new_columns = [
    "new_col STRING",
    "age_details STRUCT<adult: BOOLEAN, retirement_age: INT>",
    "hobbies ARRAY<STRING>"
]

# 拼接添加列的SQL子句
add_columns_clause = ", ".join(new_columns)

# 执行Schema变更
delta_table.alter(f"ADD COLUMNS ({add_columns_clause})")

方法2：使用Spark SQL语句

如果更习惯SQL风格，也可以动态生成ALTER TABLE语句执行：

# 同样是识别出的新增列列表
new_columns = [
    "new_col STRING",
    "age_details STRUCT<adult: BOOLEAN, retirement_age: INT>",
    "hobbies ARRAY<STRING>"
]

add_cols_str = ", ".join(new_columns)
# 执行SQL修改Schema
spark.sql(f"ALTER TABLE people_10m ADD COLUMNS ({add_cols_str})")

关键注意事项 #

1. 禁止列删除：可以通过设置表属性来阻止列删除操作，同时开启列名映射（需要Delta Lake版本支持）：

# 用DeltaTable API设置属性
delta_table.alter("SET TBLPROPERTIES ('delta.columnMapping.mode' = 'name', 'delta.minReaderVersion' = '2', 'delta.minWriterVersion' = '5')")

# 或者用SQL设置
spark.sql("ALTER TABLE people_10m SET TBLPROPERTIES ('delta.columnMapping.mode' = 'name', 'delta.minReaderVersion' = '2', 'delta.minWriterVersion' = '5')")

设置后，尝试删除列会直接报错，符合禁止删除的要求。

2. 列类型语法：嵌套列（STRUCT）和数组列（ARRAY）要遵循Spark SQL的类型定义规则，比如STRUCT要明确每个字段的名称和类型，数组要指定元素类型。

3. 动态列处理：如果新增列是动态生成的，只要确保每个列的定义格式正确（列名 类型），就能通过拼接字符串的方式批量添加，完全适配你用set API识别列的场景。

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