如何简化PySpark从JSON提取字段生成DataFrame的操作？

问题场景 #

我需要从JSON字符串中提取部分字段生成DataFrame，目前采用的是将每个字段单独转为DataFrame后再关联的方式，但实际项目中要提取的字段数量较多，这种方法过于繁琐，希望找到更高效的实现方式。

当前实现代码：

from pyspark.sql import Row
import json

s = '{"job_id":"123","settings":{"task":[{"taskname":"task1"},{"taskname":"task2"}]}}'

json_object = json.loads(s)

job_id_l = [Row(job_id=json_object['job_id'])] 
job_id_df = spark.createDataFrame(job_id_l) 

tasknames = []
for t in json_object['settings']["task"]:
  tasknames.append(Row(taskname=t["taskname"]))

tasknames_df = spark.createDataFrame(tasknames) 

job_id_df.crossJoin(tasknames_df).display()

运行结果：

job_id  taskname
123     task1
123     task2

优化方案 #

利用Spark内置的JSON解析和数组展开函数，无需手动拆分DataFrame，步骤更简洁，适配多字段场景：

from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StructField, StringType, ArrayType

# 定义JSON对应的Schema结构
json_schema = StructType([
    StructField("job_id", StringType(), nullable=True),
    StructField("settings", StructType([
        StructField("task", ArrayType(StructType([
            StructField("taskname", StringType(), nullable=True)
        ])), nullable=True)
    ]), nullable=True)
])

target_json = '{"job_id":"123","settings":{"task":[{"taskname":"task1"},{"taskname":"task2"}]}}'

# 一站式解析JSON并生成目标DataFrame
result_df = (
    spark.createDataFrame([(target_json,)], ["json_content"])
    .select(F.from_json(F.col("json_content"), json_schema).alias("parsed_data"))
    .select("parsed_data.job_id", F.explode("parsed_data.settings.task").alias("task_item"))
    .select("job_id", "task_item.taskname")
)

result_df.display()

方案优势 #

1. 结构化解析：通过定义Schema，Spark自动解析多层嵌套的JSON结构，新增字段只需在Schema中补充定义即可，无需修改业务逻辑。
2. 高效展开数组：使用explode函数直接将数组类型字段展开为多行，自动关联父级字段（如job_id），避免手动交叉关联。
3. 批量兼容：如果是处理批量JSON字符串，只需将所有字符串放入DataFrame的一列，重复上述解析流程即可，无需循环处理单个JSON对象。

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