PySpark如何将字符串类型的body列转换为新DataFrame？

问题场景 #

现有Spark DataFrame包含header和body两列，其中body列为JSON格式的字符串，需提取body中的内容，转换为包含name、age、emails列的结构化DataFrame。

错误原因分析 #

你尝试的两种方法均无法实现需求，问题出在：

1. 误用to_json函数（还拼写错误为to__json）：该函数作用是将结构化数据转为JSON字符串，与你需要的「把JSON字符串解析为结构化数据」需求完全相反；同时参数传递方式错误，sf.col无需传入类型参数，类型应通过Schema指定。
2. 直接使用select("body.*")：此时body仍是字符串类型，而非Spark可识别的Struct类型，因此无法通过.*展开字段。

正确实现方案 #

核心思路是先用from_json将JSON字符串解析为Struct类型，再提取对应字段：

步骤1：定义JSON对应的Schema

根据body的JSON结构，提前定义匹配的Schema：

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

json_schema = StructType([
    StructField("name", StringType(), nullable=True),
    StructField("age", IntegerType(), nullable=True),
    StructField("emails", ArrayType(StringType()), nullable=True)
])

步骤2：解析JSON字符串并提取字段

方法一：通过中间列分步实现

# 将body字符串解析为Struct类型
df_parsed = df.withColumn("body_struct", sf.from_json(sf.col("body"), json_schema))

# 提取Struct中的字段生成目标DataFrame
target_df = df_parsed.select("body_struct.name", "body_struct.age", "body_struct.emails")

方法二：一步到位简化写法

target_df = df.select(
    sf.from_json(sf.col("body"), json_schema).alias("body_struct")
).select("body_struct.*")

执行后即可得到符合需求的结构化DataFrame。

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