如何在Azure Databricks中更新Delta表的Map of Struct类型列值

由于transform_options是嵌套的Map<string, Struct>强类型字段，直接执行UPDATE会因类型不匹配报错，必须完整重构嵌套结构才能完成更新。以下是优先使用SQL的解决方案：

核心SQL更新语句 #

UPDATE prod.silver.control_table
SET transform_options = transform_values(
    transform_options,
    (key, val) -> struct(
        val.col_name_mappings AS col_name_mappings,
        val.type_mappings AS type_mappings,
        val.partition_duplicates_by AS partition_duplicates_by,
        array_transform(
            val.order_duplicates_by,
            elem -> CASE WHEN elem = '_commit_version' THEN 'commit_version' ELSE elem END
        ) AS order_duplicates_by
        -- 若Struct包含其他字段，需在此处原样保留，确保类型完全匹配
    )
)
-- 仅更新包含目标元素的行，避免无意义的全表更新以提升性能
WHERE EXISTS (
    SELECT 1 
    FROM UNNEST(transform_options[key].order_duplicates_by) AS elem 
    WHERE elem = '_commit_version'
);

代码说明 #

• transform_values：遍历Map的每个键值对，对Struct类型的value进行完整重构
• array_transform：逐个处理order_duplicates_by数组元素，替换指定字符串
• WHERE子句：精准过滤出需要更新的行，减少不必要的计算

低版本Runtime兼容方案 #

如果你的Databricks Runtime版本低于10.0，部分高阶函数可能不支持，可以用DataFrame API处理后写回：

from pyspark.sql.functions import col, array_transform, struct, map_from_arrays, explode, when

# 读取表并展开Map结构
df = spark.table("prod.silver.control_table") \
    .select("table_name", explode("transform_options").alias("map_key", "struct_val"))

# 更新数组中的目标元素
updated_df = df.withColumn(
    "struct_val",
    struct(
        col("struct_val.col_name_mappings"),
        col("struct_val.type_mappings"),
        col("struct_val.partition_duplicates_by"),
        array_transform(
            col("struct_val.order_duplicates_by"),
            lambda x: when(x == "_commit_version", "commit_version").otherwise(x)
        ).alias("order_duplicates_by")
    )
)

# 重新组合Map并覆盖原表
final_df = updated_df.groupBy("table_name") \
    .agg(map_from_arrays(collect_list("map_key"), collect_list("struct_val")).alias("transform_options"))

final_df.write.mode("overwrite").saveAsTable("prod.silver.control_table")

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