如何为from_json传入存储在列中的动态JSON Schema？

问题根源 #

Spark的from_json函数要求第二个参数必须是静态的Schema对象（StructType）或DDL格式的字符串字面量——因为Spark在生成执行计划阶段就需要明确JSON的结构，而col("schema")是运行时才能获取的列值，无法在计划阶段确定，所以直接传入会触发报错。

分场景解决方案 #

场景1：所有行的Schema完全相同（如你的示例数据）

如果DataFrame中schema列的所有值一致，只需先提取出Schema字符串，转换为StructType对象，再用静态Schema处理即可：

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

# 提取并补全Schema字符串（你的示例中缺少闭合的>，需先修正）
schema_str = df.select("schema").first()[0]
if not schema_str.endswith(">"):
    schema_str += ">"
# 将DDL字符串转为StructType
target_schema = StructType.fromDDL(schema_str)

# 处理数据
result_df = df.withColumn("json_data", F.explode("data")) \
              .withColumn("json_data", F.from_json(F.col("json_data"), target_schema)) \
              .select("json_data.*")

result_df.show()

场景2：不同行的Schema不同

如果schema列存在多种不同的结构，需要针对每行单独解析JSON，可使用mapInPandas（性能优于UDF）实现动态解析：

import pandas as pd
from pyspark.sql.types import StructType

def process_row_group(iterator):
    for row in iterator:
        data_array = row["data"]
        schema_str = row["schema"]
        # 修正Schema字符串格式
        if not schema_str.endswith(">"):
            schema_str += ">"
        # 转换为StructType
        target_schema = StructType.fromDDL(schema_str)
        # 解析每个JSON元素并返回
        for json_str in data_array:
            # 用pandas解析JSON并转为符合Schema的行
            pd_row = pd.read_json(pd.Series([json_str]), dtype=False).iloc[0]
            yield pd_row.to_dict()

# 生成结果DataFrame，这里以第一行的Schema作为输出结构基准（若Schema多样，可根据实际情况调整）
output_schema = StructType.fromDDL(df.select("schema").first()[0])
result_df = df.mapInPandas(process_row_group, schema=output_schema)

result_df.show()

注意事项 #

• 确保schema列的DDL字符串格式正确，必须包含闭合的>，否则StructType.fromDDL会解析失败。
• 若存在多种Schema，输出DataFrame的结构会以指定的output_schema为准，若不同Schema的字段差异较大，需额外处理字段兼容问题。

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