使用PySpark解析CSV中JSON列提取ServiceType字段遇空值求解

针对你遇到的additional_json列首尾带引号、内部有反斜杠转义的情况，可以通过两步处理：先清理字符串格式，再解析JSON提取目标字段。

方法一：先清理字符串，再完整解析JSON #

1. 清理格式：用正则表达式去掉首尾的引号，同时移除所有反斜杠转义符
2. 解析JSON：用from_json将清理后的字符串转为结构体，再提取ServiceType

示例代码：

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, regexp_replace, from_json
from pyspark.sql.types import StructType, StructField, StringType

# 初始化SparkSession
spark = SparkSession.builder.appName("CleanJsonColumn").getOrCreate()

# 模拟测试数据
data = [("{\"ServiceType\":\"Delivery\",\"OrderId\":\"123\"}",),
        ("{\"ServiceType\":\"Pickup\",\"OrderId\":\"456\"}",)]
df = spark.createDataFrame(data, ["additional_json"])

# 清理字符串：去掉首尾引号 + 移除反斜杠
cleaned_df = df.withColumn(
    "cleaned_json",
    regexp_replace(
        regexp_replace(col("additional_json"), "^\"|\"$", ""),  # 匹配首尾引号并删除
        "\\\\", ""  # 移除所有反斜杠
    )
)

# 定义JSON schema，只保留需要的ServiceType字段
json_schema = StructType([StructField("ServiceType", StringType())])

# 解析JSON并提取ServiceType
result_df = cleaned_df.withColumn(
    "service_type",
    from_json(col("cleaned_json"), json_schema).getField("ServiceType")
)

result_df.select("additional_json", "service_type").show(truncate=False)

方法二：直接提取目标字段（无需完整解析JSON） #

如果只需要ServiceType这一个字段，可以用regexp_extract直接匹配取值，效率更高：

from pyspark.sql.functions import regexp_extract

result_df = df.withColumn(
    "service_type",
    regexp_extract(
        col("additional_json"),
        r'"ServiceType":"([^"]+)"',  # 正则匹配ServiceType的值
        1  # 提取第一个分组的内容
    )
)

result_df.select("additional_json", "service_type").show(truncate=False)

两种方法都能解决你的问题，方法一适合需要提取多个JSON字段的场景，方法二更适合只需要单个字段的情况。

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