如何在Spark自连接中识别Databricks Delta表的所有变更列？

方法一：使用动态SQL（Databricks SQL） #

由于表中列数量多，手动逐个写比较逻辑不现实，可借助元数据获取非键列，再通过动态SQL检查每个列是否存在变更：

1. 获取所有非键列
   先从元数据中筛选出排除键列后的目标列：

SELECT column_name
FROM information_schema.columns
WHERE table_name = 'table1' -- 替换为你的表名
  AND column_name NOT IN ('Key1', 'Key2', 'Key3')

2. 动态生成变更检查查询
   利用上述列列表，通过EXECUTE IMMEDIATE执行动态拼接的SQL，找出所有存在变更的列：

DECLARE cols STRING;
SET cols = (
  SELECT STRING_AGG(column_name, ', ')
  FROM information_schema.columns
  WHERE table_name = 'table1'
    AND column_name NOT IN ('Key1', 'Key2', 'Key3')
);

EXECUTE IMMEDIATE CONCAT(
  'WITH column_checks AS (',
  (SELECT STRING_AGG(
    CONCAT(
      'SELECT "', column_name, '" AS column_name FROM table1 t1 JOIN table1 t2 ON t1.Key1 = t2.Key1 AND t1.Key2 = t2.Key2 AND t1.Key3 = t2.Key3 WHERE t1.', column_name, ' IS DISTINCT FROM t2.', column_name, ' LIMIT 1'
    ),
    ' UNION ALL '
  ) FROM information_schema.columns WHERE table_name = 'table1' AND column_name NOT IN ('Key1', 'Key2', 'Key3')),
  ') SELECT DISTINCT column_name FROM column_checks'
)

这里用IS DISTINCT FROM替代!=，能正确处理NULL值的差异（比如一方为NULL、另一方有值的情况）。

方法二：使用PySpark（编程式处理） #

如果熟悉PySpark，这种方式更灵活，适合自动化处理：

# 获取表数据
df = spark.table("table1")

# 定义键列和非键列
key_cols = ["Key1", "Key2", "Key3"]
non_key_cols = [col for col in df.columns if col not in key_cols]

# 遍历检查每个非键列是否存在变更
changed_cols = []
for col in non_key_cols:
    # 检查是否存在该列值不同的关联记录
    has_change = df.alias("t1") \
        .join(df.alias("t2"), on=key_cols) \
        .filter(f"t1.{col} IS DISTINCT FROM t2.{col}") \
        .limit(1) \
        .count() > 0
    if has_change:
        changed_cols.append(col)

# 输出结果
print("发生过变更的列：")
for col in changed_cols:
    print(col)

注意事项 #

• 若表数据量极大，直接全表自连接可能性能不佳，可考虑先对每个键分组取最新和次新的记录（如果日志包含时间戳字段），再做比较，减少数据处理量。
• IS DISTINCT FROM是ANSI SQL标准函数，Databricks完全支持，比!=更准确，避免NULL值导致的漏判。

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