如何使用PySpark解析Blob格式嵌套JSON并补全缺失字段

核心解决思路 #

针对结构不一致的JSON记录，最可靠的方式是提前定义包含所有可能字段的Schema，让PySpark读取时自动为缺失字段填充Null；若已完成数据读取，也可通过动态添加字段的方式补全缺失值。

步骤1：读取Blob中的换行分隔JSON数据 #

直接使用PySpark的JSON读取器，指定换行符分隔模式。若Blob存储（如Azure Blob、AWS S3）需要认证，需提前配置对应存储的访问凭证。

# 以Azure Blob为例，读取换行分隔的JSON数据
df = spark.read.format("json")\
    .option("lineSep", "\n")\
    .load("wasbs://<容器名>@<存储账户>.blob.core.windows.net/<数据路径>/")

步骤2：显式定义Schema（推荐方案） #

自动推断Schema可能因字段缺失导致结构异常，手动定义覆盖所有可能字段的Schema，能确保缺失字段统一填充为Null。

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

# 根据实际业务场景，定义包含所有可能字段的Schema
custom_schema = StructType([
    StructField("EventId", StringType(), nullable=True),
    StructField("EventType", StringType(), nullable=True),
    StructField("Channel Id", StringType(), nullable=True),
    StructField("Conversation Id", StringType(), nullable=True),
    StructField("Timestamp", TimestampType(), nullable=True),
    StructField("UserID", StringType(), nullable=True),
    StructField("EventValue", IntegerType(), nullable=True)
])

# 读取数据时指定自定义Schema
df = spark.read.format("json")\
    .option("lineSep", "\n")\
    .schema(custom_schema)\
    .load("wasbs://<容器名>@<存储账户>.blob.core.windows.net/<数据路径>/")

步骤3：已读取数据后补全缺失字段 #

如果已经完成数据读取，可通过遍历目标字段列表，动态添加缺失字段并赋值为Null。

from pyspark.sql.functions import lit
from pyspark.sql.types import StringType

# 定义需要确保存在的字段列表
required_fields = ["Channel Id", "Conversation Id"]

for field in required_fields:
    if field not in df.columns:
        # 根据字段实际类型调整cast的类型
        df = df.withColumn(field, lit(None).cast(StringType()))

步骤4：验证结果 #

查看DataFrame的结构和数据，确认缺失字段已正确填充为Null：

# 打印Schema结构，检查字段是否齐全
df.printSchema()

# 查看数据内容，验证缺失值填充情况
df.show(truncate=False)

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