如何用PySpark内置函数替代UDF处理转义JSON格式列

问题分析 #

你的原始列值是被双重转义的JSON字符串：外层包裹双引号，内部JSON的键/值引号被转义为两个连续双引号。目标是将其转换为合法的JSON格式（或可直接解析的结构化数据），同时避免多次调用UDF带来的性能损耗。

解决方案：用PySpark内置函数实现 #

完全可以通过PySpark内置的字符串处理+JSON解析函数实现，无需UDF，性能会大幅提升。具体步骤如下：

1. 清理字符串，得到合法JSON格式 #

首先去掉外层包裹的双引号，再将内部的""替换为"，得到标准JSON字符串：

from pyspark.sql import functions as F

# 假设待处理列名为raw_json_col
df = df.withColumn(
    "cleaned_json_str",
    # 1. 去掉首尾的双引号：从第2个字符取到倒数第2个字符
    # 2. 将所有""替换为"
    F.regexp_replace(
        F.substring(F.col("raw_json_col"), 2, F.length(F.col("raw_json_col")) - 2),
        '""',
        '"'
    )
)

2. 解析为结构化数据（可选但推荐） #

如果后续需要对JSON内的字段进行操作，可直接将清理后的字符串解析为MapType或自定义StructType：

from pyspark.sql.types import MapType, StringType, DoubleType

# 解析为Map<String, Double>类型
df = df.withColumn(
    "parsed_json",
    F.from_json(F.col("cleaned_json_str"), MapType(StringType(), DoubleType()))
)

# 若需要解析为指定结构体（比如明确字段），可定义StructType：
# from pyspark.sql.types import StructType, StructField
# json_schema = StructType([
#     StructField("ab", DoubleType()),
#     StructField("cd", DoubleType()),
#     StructField("ef", DoubleType()),
#     StructField("gh", DoubleType()),
#     StructField("ij", DoubleType()),
#     StructField("kl", DoubleType())
# ])
# df = df.withColumn("parsed_struct", F.from_json(F.col("cleaned_json_str"), json_schema))

性能优势 #

PySpark内置函数在JVM层面执行，避免了Python UDF需要的Python-JVM序列化/反序列化开销，尤其在处理多列、大数据量时，性能提升非常明显，无需为每个列单独编写UDF。

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