数据质量检查：PySpark对比String类型列生成差异列的方法

需求说明 #

需对比两个存储数组格式字符串的StringType列（old_unmatch、current_unmatch），生成两个新列：

• new_unmatch：current_unmatch相较于old_unmatch的新增元素
• missed_unmatch：old_unmatch相较于current_unmatch的缺失元素

示例效果 #

实现方案（以PySpark为例） #

1. 将StringType数组字符串转换为ArrayType

原列是字符串格式的数组（如"['121', '122']"），需先解析为Spark的ArrayType类型：

from pyspark.sql import functions as F

# 提取字符串中的元素，转换为数组
df = df.withColumn("old_array", F.regexp_extract_all(F.col("old_unmatch"), r"'(.*?)'", 1))
df = df.withColumn("current_array", F.regexp_extract_all(F.col("current_unmatch"), r"'(.*?)'", 1))

2. 计算新增与缺失元素

利用array_except函数计算两个数组的差集，直接得到目标列：

# 计算current相对于old的新增元素
df = df.withColumn("new_unmatch", F.array_except(F.col("current_array"), F.col("old_array")))
# 计算old相对于current的缺失元素
df = df.withColumn("missed_unmatch", F.array_except(F.col("old_array"), F.col("current_array")))

3. 可选：将结果转回StringType格式

若需要保持与原列一致的字符串数组格式，可将结果转换回去：

# 格式化为['元素1', '元素2']样式的字符串
df = df.withColumn(
    "new_unmatch",
    F.concat(F.lit("["), F.array_join(F.col("new_unmatch"), "', '"), F.lit("']"))
)
df = df.withColumn(
    "missed_unmatch",
    F.concat(F.lit("["), F.array_join(F.col("missed_unmatch"), "', '"), F.lit("']"))
)

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