PySpark：检测DataFrame列中缺闭合大括号的字典值并提取内容

一、正确筛选格式错误的字典行 #

你用df = df.where(~df.column.contains('}'))的逻辑是对的，若只看到部分值，大概率是PySpark默认显示长度限制导致内容被截断。先设置参数让完整内容显示：

from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()
spark.conf.set("spark.sql.repl.eagerEval.truncate", -1)  # 关闭内容截断，显示完整字符串

# 重新筛选查看完整结果
bad_rows_df = df.where(~df["column"].contains('}'))
bad_rows_df.show()

如果列不是字符串类型，先转成字符串再筛选：

from pyspark.sql.functions import col

bad_rows_df = df.where(~col("column").cast("string").contains('}'))

二、提取错误字典的末尾内容 #

针对{'Id': '1', 'state': 'FL', "zip": "3333"这类缺闭合大括号的脏数据，用regexp_extract结合正则提取最后字段的值，兼容混合单双引号的情况：

from pyspark.sql.functions import regexp_extract

bad_rows_df = bad_rows_df.withColumn(
    "last_value",
    regexp_extract(
        col("column"),
        r':\s*([\'"])(.*?)\1\s*$',  # 匹配最后一对引号包裹的内容
        2  # 提取捕获组的目标值
    )
)

bad_rows_df.select("column", "last_value").show()

若最后字段无引号包裹，调整正则为匹配冒号到行尾的内容：

bad_rows_df = bad_rows_df.withColumn(
    "last_value",
    regexp_extract(
        col("column"),
        r':\s*(.*?)\s*$',
        1
    )
)

三、标记正确/错误字典行 #

新增标识列区分格式状态：

from pyspark.sql.functions import when

df = df.withColumn(
    "is_valid_dict",
    when(
        col("column").cast("string").contains('{') & col("column").cast("string").contains('}'),
        True
    ).otherwise(False)
)

df.select("column", "is_valid_dict").show()

进阶严谨校验：用try_cast尝试转成Map类型，解析失败的即为格式错误行：

from pyspark.sql.functions import try_cast

df = df.withColumn(
    "parsed_dict",
    try_cast(col("column").cast("string"), "map<string, string>")
)

bad_rows_df = df.where(col("parsed_dict").isNull())

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