PySpark：基于Schema列将JSON字符串列转换为JSON结构体列

你的代码不生效的核心原因是：from_json函数的第二个参数要求传入静态的StructType对象，而你直接传入了DataFrame中的JsonSchema列（存储的是字符串格式的schema），Spark无法将列对象识别为合法的schema定义，因此解析失败。

下面分两种场景给出实现方案：

场景1：所有行的JsonSchema完全一致（全局统一schema） #

如果你的DataFrame中所有行的JsonSchema列内容相同，只需先将字符串schema解析为StructType对象，再传入from_json即可：

import json
from pyspark.sql import functions as F
from pyspark.sql.types import StructType

# 取出任意一行的schema字符串（假设所有行schema一致）
schema_str = df.select("JsonSchema").first()[0]
# 将字符串schema解析为StructType对象
json_schema = StructType.fromJson(json.loads(schema_str))

# 执行JSON解析
df1 = df.withColumn("NewJson", F.from_json(F.col("JsonData"), json_schema))

场景2：每行的JsonSchema各不相同（行级动态schema） #

如果每行对应不同的schema定义，Spark原生的from_json无法直接支持，需要通过Python UDF实现动态解析：

import json
from pyspark.sql import functions as F
from pyspark.sql.types import StructType, Row, AnyType

# 定义自定义UDF，接收json字符串和schema字符串，返回解析后的结构体
def parse_dynamic_json(json_data, schema_str):
    try:
        # 解析schema字符串为StructType
        schema = StructType.fromJson(json.loads(schema_str))
        # 解析json数据为字典
        parsed_dict = json.loads(json_data)
        # 按照schema字段构建Row对象
        return Row(**{field.name: parsed_dict.get(field.name) for field in schema.fields})
    except Exception:
        # 处理解析失败的情况，返回None或自定义默认值
        return None

# 注册UDF，返回类型设为AnyType（因为每行返回结构不同）
dynamic_parse_udf = F.udf(parse_dynamic_json, AnyType())

# 应用UDF生成新列
df1 = df.withColumn("NewJson", dynamic_parse_udf(F.col("JsonData"), F.col("JsonSchema")))

⚠️ 注意：使用AnyType会导致Spark无法推断列的结构，后续对NewJson列进行SQL操作（如字段提取）会受限，这种场景建议尽量避免，优先统一schema。

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