如何为Delta表列动态添加验证及生成缺失属性列？

场景回顾 #

读取的原始Delta表数据：

+--------+------------------+
| emp_id |     str          |
+--------+------------------+
|   1    | name=qwerty.     |
|   2    | age=22           | 
|   3    | job=googling     | 
|   4    | dob=12-Jan-2001  | 
|   5    | weight=62.7.     | 
+--------+------------------+

拆分str列后生成的宽表（无预定义Schema，缺失列填充null）：

+--------+--------+------+----------+-------------+--------+
| emp_id |  name  |  age |    job   |     dob     | weight |
+--------+--------+------+----------+-------------+--------+
|   1    | qwerty | null |   null   |     null    |  null  |
|   2    |  null  |  22  |   null   |     null    |  null  | 
|   3    |  null  | null | googling |     null    |  null  | 
|   4    |  null  | null |   null   | 12-Jan-2001 |  null  | 
|   5    |  null  | null |   null   |     null    |  62.7  | 
+--------+--------+------+----------+-------------+--------+

问题1：能否在步骤2中基于列名进行验证？还是必须在步骤3处理新DataFrame时再做验证？ #

可以在步骤2（拆分生成宽表的阶段）直接完成列名验证，无需等到步骤3。

实现思路

1. 先定义允许的目标列集合：["name", "age", "job", "dob", "weight"]
2. 拆分str列提取key和value时，直接过滤掉不在允许集合内的key；也可以对无效key做单独标记（比如存入invalid_attributes列）
3. 生成宽表时只保留允许的列，从根源上避免无效列进入后续流程

代码示例（PySpark）

from pyspark.sql import functions as F

# 定义允许的列
allowed_cols = ["name", "age", "job", "dob", "weight"]

# 读取原始Delta表
df = spark.read.format("delta").load("path/to/your/delta_table")

# 拆分str列，提取key和value，同时清洗value（去掉末尾的点）
split_df = df.withColumn("split_str", F.split(F.col("str"), "=")) \
             .withColumn("key", F.trim(F.col("split_str")[0])) \
             .withColumn("value", F.trim(F.regexp_replace(F.col("split_str")[1], r"\.$", ""))) \
             .filter(F.col("key").isin(allowed_cols))  # 过滤无效列

# 转成宽表
wide_df = split_df.groupBy("emp_id") \
                  .pivot("key", allowed_cols) \
                  .agg(F.first("value"))

问题2：能否生成包含missing_attributes列的Delta表？ #

完全可以，通过Spark内置函数即可实现，无需复杂逻辑。

实现思路

1. 确定所有需要检查的目标列（即除emp_id外的列）
2. 对每一行，筛选出值为null的列名，用逗号拼接成字符串作为missing_attributes列

代码示例（PySpark）

# 基于问题1生成的wide_df，获取目标列列表
target_cols = [col for col in wide_df.columns if col != "emp_id"]

# 构造missing_attributes列：过滤出值为null的列名，拼接成字符串
result_df = wide_df.withColumn(
    "missing_attributes",
    F.concat_ws(
        ",",
        F.filter(
            F.array(*[F.when(F.col(c).isNull(), F.lit(c)) for c in target_cols]),
            lambda x: x.isNotNull()
        )
    )
)

# 写入Delta表
result_df.write.format("delta").mode("overwrite").save("path/to/target_delta_table")

生成的结果表如下：

+--------+--------+------+----------+-------------+--------+---------------------+
| emp_id |  name  |  age |    job   |     dob     | weight | missing_attributes  |
+--------+--------+------+----------+-------------+--------+---------------------+
|   1    | qwerty | null |   null   |     null    |  null  | age,job,dob,weight  |
|   2    |  null  |  22  |   null   |     null    |  null  | name,job,dob,weight | 
|   3    |  null  | null | googling |     null    |  null  | name,age,dob,weight |
|   4    |  null  | null |   null   | 12-Jan-2001 |  null  | name,age,job,weight |
|   5    |  null  | null |   null   |     null    |  62.7  | name,age,job,dob    |
+--------+--------+------+----------+-------------+--------+---------------------+

内容的提问来源于stack exchange，提问作者boring-coder