PySpark创建含JSON列DataFrame遇类型错误及后续方案咨询

一、初始构建的错误原因与修复 #

错误原因 #

你定义的schema中，groups是**ArrayType(StructType)**类型，但传入的数据data = [groups_rows]里，每个元素是一个包含groups键的字典。Spark期望的是直接传入groups字段对应的值（也就是那个数组）——因为schema已经指定了列名为groups，不需要再在外层套一层字典。

正确构建方式 #

直接把数组作为DataFrame的行数据，去掉外层的字典包裹：

from pyspark.sql import types as T
from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()

# 定义schema
schema = T.StructType([
    T.StructField(
        'groups', T.ArrayType(
            T.StructType([
                T.StructField("types", T.ArrayType(T.StringType(), False)),
                T.StructField("label", T.StringType())
            ])
        )
    )
])

# 准备数据：直接传入groups字段对应的数组值
groups_data = [
    [
        {"types": ["baseball", "basketball"], "label": "Label 1"},
        {"types": ["football"], "label": "Label 2"}
    ]
]

sections_df = spark.createDataFrame(data=groups_data, schema=schema)
sections_df.show(truncate=False)

是否需要用MapType？ #

不需要。MapType适用于键值对不固定的动态场景，而你的groups元素是固定结构（有types和label两个固定字段），用StructType能保证数据结构一致性，后续字段操作也更方便。

二、JSON字符串转换方案的错误修复 #

错误原因 #

1. Schema定义错误：你把StructType误写成了MapType，MapType的参数是键类型和值类型，不是结构体字段，这里必须用ArrayType(StructType)。
2. 函数使用错误：你用反了to_json和from_json——to_json是把结构化数据转成JSON字符串，而from_json才是把JSON字符串解析成结构化数据；另外代码里还引用了不存在的sections列。

正确实现方式 #

如果要先构造JSON字符串再解析成结构化列，代码如下：

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

# 原始数据
groups = [
    {"recTypes": ["readinghistory", "popular"], "sectionLabel": "Reader favorites you missed"},
    {"recTypes": ["contentpacks"], "sectionLabel": "Based on your interests"}
]

# 转成JSON字符串
groups_json = json.dumps(groups)

# 正确定义schema：字段名要和JSON里的键完全对应
groups_schema = T.ArrayType(
    T.StructType([
        T.StructField("recTypes", T.ArrayType(T.StringType(), False)),
        T.StructField("sectionLabel", T.StringType())
    ])
)

# 创建基础DataFrame，再添加解析后的groups列
df = spark.createDataFrame([(1,)], ["id"])
df = df.withColumn("groups", F.lit(groups_json)) \
       .withColumn("groups", F.from_json(F.col("groups"), groups_schema))

df.show(truncate=False)

如果只需要groups一列，可以简化为：

df = spark.createDataFrame([(groups_json,)], ["groups_str"]) \
       .withColumn("groups", F.from_json(F.col("groups_str"), groups_schema)) \
       .drop("groups_str")

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