PySpark定义Rest API响应Schema时遭遇ValueError问题求助

错误根源 #

1. Schema定义不匹配：原API返回的metadata是嵌套结构体（包含count字段），但你将其定义为StringType，导致解析后的结构化数据与Schema类型冲突。
2. UDF返回类型错误：如果UDF返回的是tuple而非Row对象，Spark无法将tuple映射到StructType，从而抛出"Unexpected tuple with StructType"错误。

修正后的Schema定义 #

根据API返回的JSON结构，调整Schema以匹配实际数据：

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

# 完全匹配API响应的Schema（按需保留字段）
schema = StructType([
    StructField("metadata", StructType([
        StructField("count", IntegerType(), nullable=True)
    ]), nullable=True),
    StructField("payload", ArrayType(
        StructType([
            StructField("id", IntegerType(), nullable=True),
            StructField("id1", IntegerType(), nullable=True)
            # 若需要其他字段（如id2、year），可在此添加对应StructField
        ])
    ), nullable=True)
])

正确的UDF写法 #

UDF必须返回Row对象以匹配StructType结构，示例如下：

from pyspark.sql.functions import udf
from pyspark.sql import Row
import json

@udf(schema=schema)
def parse_api_response(json_str):
    data = json.loads(json_str)
    # 构造metadata对应的Row
    metadata = Row(count=int(data["metadata"]["count"]))
    # 构造payload数组中的每个元素Row
    payload = [
        Row(id=int(item["id"]), id1=int(item["id1"])) 
        for item in data["payload"]
    ]
    # 返回顶层Row，匹配Schema结构
    return Row(metadata=metadata, payload=payload)

更高效的替代方案：无需UDF #

如果只是解析JSON字符串为结构化数据，推荐使用Spark内置的from_json函数，比UDF更高效：

from pyspark.sql.functions import from_json, col

# 假设你的DataFrame包含存储API响应JSON的列`api_response`
df = df.withColumn("parsed_data", from_json(col("api_response"), schema))

# 提取解析后的字段
df.select(
    "parsed_data.metadata.count",
    "parsed_data.payload.id",
    "parsed_data.payload.id1"
).show()

内容的提问来源于stack exchange，提问作者Niladri Das Choudhury