PySpark新手用StructType指定Schema创建DataFrame遇类型错误

错误原因 #

spark.createDataFrame()的第一个参数要求是可迭代的行数据集合（比如列表、RDD等），你直接传入了单个字典对象jsn，Spark会将字典的键（如"books"）当作独立元素处理，这和你定义的StructType Schema结构不匹配，因此抛出类型错误。

解决方案 #

方法1：将单个字典包装为列表 #

把你的字典放在列表中，让createDataFrame识别这是一行数据：

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

jsn = {"books": [{"book_id": "B001","book_name": "The Great Gatsby","author": "F. Scott Fitzgerald","copies_available": 5},{"book_id": "B002","book_name": "To Kill a Mockingbird","author": "Harper Lee","copies_available": 3}]}

js_schema = StructType([
    StructField("books",ArrayType(StructType([
        StructField("book_id",StringType()),
        StructField("book_name",StringType()),
        StructField("author",StringType()),
        StructField("copies_available",IntegerType())
    ])))
])

# 用列表包裹字典，作为行数据传入
df = spark.createDataFrame([jsn], js_schema)
df.show(truncate=False)

运行后会输出正确的DataFrame：

+------------------------------------------------------------------------------------------------------------------------------------+
|books                                                                                                                               |
+------------------------------------------------------------------------------------------------------------------------------------+
|[{B001, The Great Gatsby, F. Scott Fitzgerald, 5}, {B002, To Kill a Mockingbird, Harper Lee, 3}]|
+------------------------------------------------------------------------------------------------------------------------------------+

方法2：使用Spark JSON读取器（推荐） #

如果你的JSON数据来自字符串或文件，更推荐用spark.read.schema()方法加载，这是PySpark处理JSON数据的标准方式：

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

js_schema = StructType([
    StructField("books",ArrayType(StructType([
        StructField("book_id",StringType()),
        StructField("book_name",StringType()),
        StructField("author",StringType()),
        StructField("copies_available",IntegerType())
    ])))
])

# 从JSON字符串加载
json_str = """{"books": [{"book_id": "B001","book_name": "The Great Gatsby","author": "F. Scott Fitzgerald","copies_available": 5},{"book_id": "B002","book_name": "To Kill a Mockingbird","author": "Harper Lee","copies_available": 3}]}"""
df = spark.read.schema(js_schema).json(sc.parallelize([json_str]))

# 若从文件加载，替换为文件路径
# df = spark.read.schema(js_schema).json("/path/to/your/json/file.json")

df.printSchema()
df.show(truncate=False)

这种方式无需手动处理行数据结构，Spark会自动按照指定Schema解析JSON内容。

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