使用Python处理多JSON转Delta时遇AnalysisException错误求助

尝试从文件夹读取多个JSON文件，通过DataFrame执行explode操作后写入Delta格式时，触发报错：

AnalysisException: Can only star expand struct data types. Attribute: ArrayBuffer(Result)

示例JSON片段 #

{"result": [{"approval": "Approved", "assigned_to": "", "assignment_group": "Business Requests", "opened_at": "08/10/2016 08:36:35", "opened_by": "Kay", "priority": "4 - Low", "sys_created_by": "p1843591@production.local", "sys_created_on": "08/10/2016 08:36:35", "urgency": "3 - Low"}, {"approval": "Approved", "assigned_to": "", "assignment_group": "Business Requests", "opened_at": "08/10/2016 10:13:11", "opened_by": "Alan", "priority": "4 - Low", "sys_created_on": "08/10/2016 10:13:11", "urgency": "3 - Low"}]}

使用的PySpark代码 #

from pyspark.sql.functions import *
import glob
spark.conf.set("spark.sql.legacy.json.allowEmptyString.enabled", True)
for file in glob.glob(f"/dbfs/mnt/landing/item/2023*/*.json"):
    file = file[5:]
    df = spark.read.option('multiline',True).json(file)
    df_exp = df.select(explode(col("result")).alias('Result')).select('Result.*')
    df_explode = df_exp.drop_duplicates()
    df_explode.write.format('delta').mode('append').option('inferSchema',True).save('/mnt/landing/finalitem/')

问题原因 #

报错说明Result不是Struct类型，无法用*展开，常见诱因：

• 部分JSON文件中result字段为空数组，导致explode后生成的Result为Null或非Struct类型
• Spark自动推断Schema时出现异常，result字段未被正确识别为数组嵌套Struct类型

解决方法 #

1. 手动指定Schema，避免推断错误 #

提前定义与JSON结构匹配的Schema，强制Spark识别正确的字段类型：

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

# 定义精准匹配的Schema
item_schema = StructType([
    StructField("result", ArrayType(StructType([
        StructField("approval", StringType()),
        StructField("assigned_to", StringType()),
        StructField("assignment_group", StringType()),
        StructField("opened_at", StringType()),
        StructField("opened_by", StringType()),
        StructField("priority", StringType()),
        StructField("sys_created_by", StringType()),
        StructField("sys_created_on", StringType()),
        StructField("urgency", StringType())
    ])))
])

# 读取时指定Schema
df = spark.read.option('multiline',True).schema(item_schema).json(file)

2. 过滤无效数据后再展开 #

在explode操作后过滤掉Result为Null的行，确保只有合法Struct数据被展开：

df_exp = df.select(explode(col("result")).alias('Result')).filter(col("Result").isNotNull())
df_exp = df_exp.select('Result.*')

3. 优化文件读取方式 #

避免循环读取单个文件，直接批量读取匹配路径的所有文件，提升效率同时减少Schema推断异常：

# 批量读取所有目标文件
df = spark.read.option('multiline',True).schema(item_schema).json("/dbfs/mnt/landing/item/2023*/*.json")
df_exp = df.select(explode(col("result")).alias('Result')).filter(col("Result").isNotNull()).select('Result.*')
df_explode = df_exp.drop_duplicates()
df_explode.write.format('delta').mode('append').save('/mnt/landing/finalitem/')

4. 移除写入时的Schema推断选项 #

Delta Lake要求追加数据的Schema与已有表Schema一致，去掉inferSchema选项避免Schema冲突：

df_explode.write.format('delta').mode('append').save('/mnt/landing/finalitem/')

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