使用Databricks Auto Loader自动推断Base64编码JSON字段的Schema

Auto Loader只会默认识别顶层的offset和value字段，value作为Base64字符串自然不会自动解析内部JSON结构。要搞定这个场景，得手动解码Base64，再结合Spark的动态Schema推断能力适配可变结构。

1. 读取原始数据并解码Base64 #

先通过Auto Loader读取Blob存储里的JSON，此时value是字符串类型，接着用Spark内置函数解码Base64并转成可读的字符串：

from pyspark.sql.functions import unbase64, col, from_json

# 读取原始流数据
raw_stream_df = (spark.readStream
                 .format("cloudFiles")
                 .option("cloudFiles.format", "json")
                 .option("cloudFiles.schemaLocation", "/dbfs/path/to/schema-store")  # 保存顶层Schema演进记录
                 .load("abfss://container@storageaccount.dfs.core.windows.net/path/to/files"))

# 解码Base64字段为字符串
decoded_df = raw_stream_df.withColumn("value_decoded", unbase64(col("value")).cast("string"))

2. 解析动态JSON结构 #

因为内部JSON的Schema是动态可变的，不能提前固定定义，这里提供两种实用方案：

方案A：自动推断内部Schema（适合初始加载或Schema变化不频繁）

先从数据中采样推断内部Schema，再用这个Schema解析全量数据：

# 从静态采样数据推断内部Schema（流处理场景可先读一批静态文件）
sample_raw_df = spark.read.json("abfss://container@storageaccount.dfs.core.windows.net/path/to/sample-files")
sample_decoded_rdd = sample_raw_df.select(unbase64(col("value")).cast("string")).rdd.map(lambda row: row[0])
internal_schema = spark.read.json(sample_decoded_rdd).schema

# 解析所有数据中的内部JSON
parsed_df = decoded_df.withColumn("value_json", from_json(col("value_decoded"), internal_schema))

# 流输出时开启Schema合并，适配后续Schema变化
query = (parsed_df.writeStream
         .format("delta")
         .option("mergeSchema", "true")
         .option("checkpointLocation", "/dbfs/path/to/checkpoint")
         .start("/dbfs/path/to/output-table"))

方案B：Schema提示+自动合并（适合有核心固定字段的动态场景）

如果内部JSON有固定的核心字段，可先定义Schema提示，让Spark自动推断并合并其他新增字段：

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

# 定义核心字段的Schema提示
core_schema = StructType([
    StructField("user_id", IntegerType(), nullable=True),
    StructField("event_time", StringType(), nullable=True)
])

# 解析时指定提示Schema，开启自动合并
parsed_df = decoded_df.withColumn(
    "value_json",
    from_json(
        col("value_decoded"),
        core_schema,
        options={"mergeSchema": "true"}
    )
)

关键注意点 #

• 必须配置cloudFiles.schemaLocation，Auto Loader会自动维护顶层Schema的演进，避免重复推断出错
• 流处理场景下务必开启mergeSchema选项，否则后续Schema变化时会抛出异常
• 解码后的Unicode字符串无需额外处理，Spark的from_json函数会自动识别并解析

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