如何用Python字典和SQL UPDATE更新Delta表中MAP类型列的值

正确实现代码 #

要正确更新MAP类型列，需要将Python字典转换成符合Spark SQL map()函数语法的参数，并把datetime对象转为Spark可识别的TIMESTAMP格式：

from datetime import datetime

# 定义时间格式化函数，转为Spark TIMESTAMP支持的字符串格式（保留三位毫秒）
def format_ts(dt: datetime) -> str:
    return dt.strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]

table_updates_id1 = {
    'id1_table_1': datetime(2023, 3, 26, 4, 33, 22, 323000),
    'id1_table_2': datetime(2023, 3, 26, 4, 33, 22, 323000)
}

# 生成map函数的参数列表：每个键用单引号包裹，时间转为timestamp()函数调用
map_args = []
for key, ts in table_updates_id1.items():
    map_args.append(f"'{key}'")
    map_args.append(f"timestamp('{format_ts(ts)}')")

# 拼接成完整的map表达式
map_expression = ', '.join(map_args)

# 执行UPDATE语句
spark.sql(f"""
UPDATE dev.bronze.test_map
SET table_updates = map({map_expression})
WHERE id = 1
""")

之前尝试失败的原因 #

1. 第一次尝试错误：直接将Python字典{table_updates_id1}代入SQL，Spark SQL的map()函数语法要求是map(key1, value1, key2, value2...)，而非Python字典格式，因此触发语法错误。
2. 第二次尝试错误：字符串拼接逻辑错误，生成的参数格式不符合要求——既没有给字符串键正确添加单引号，也没有将datetime对象转为Spark可识别的TIMESTAMP类型，导致SQL解析失败。

验证更新结果 #

执行完UPDATE后，可以用以下SQL验证结果：

SELECT id, table_updates FROM dev.bronze.test_map WHERE id = 1;

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