PySpark处理CSV中缺失闭合大括号的JSON字段问题

报错原因解释 #

你用to_json报错是因为这个函数的作用是把Spark结构化数据（如struct、map、数组类型）转换为JSON字符串，而你的col1本身就是字符串类型，类型不匹配导致报错，这个函数不适合用来处理已有的字符串格式JSON。

解决方案步骤 #

我们需要分两步实现需求：标记格式错误的行，提取错误行的最后一个字段值。

1. 导入必要模块 #

from pyspark.sql import functions as F

2. 标记缺失闭合大括号的行 #

可以通过两种方式识别错误行：

# 方式1：判断行尾不是}（简单直接，适配你的示例场景）
df = df.withColumn(
    "is_invalid",
    F.col("col1").rlike(r"[^}]$")
)

# 方式2：统计{和}的数量，差为1则说明缺一个闭合大括号（更严谨）
df = df.withColumn(
    "is_invalid",
    (F.length(F.regexp_replace("col1", "[^{]", "")) - F.length(F.regexp_replace("col1", "[^}]", ""))) == 1
)

3. 提取错误行的最后一个字段值 #

针对错误行，用正则提取最后一个键值对的值（以示例中的business键为例）：

df = df.withColumn(
    "last_value",
    F.when(
        F.col("is_invalid"),
        # 匹配'business:'后的内容直到行尾，提取捕获组
        F.regexp_extract("col1", r"'business':\s*(.*)$", 1)
    ).otherwise(None)
)

4. 查看结果 #

df.select("col1", "is_invalid", "last_value").show(truncate=False)

示例输出说明 #

对于你提供的错误行：

"{'Id': '1', 'fruit': ['banana', 'grape', 'orange'], 'business': ""Publix""

is_invalid会标记为True，last_value会提取出""Publix""；正确行则is_invalid为False，last_value为null。

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