Event Hub捕获Avro文件后，Databricks Autoloader能否启用Schema Evolution？

首先明确：外层的Event Hub Capture Avro Schema不会直接导致无法使用Schema Evolution，但Autoloader默认只会识别外层Avro的固定Schema（包含Message Body、Offset等字段），而你的业务Schema嵌套在Message Body的JSON里，所以默认情况下Autoloader无法感知业务Schema的变化。下面是具体的解决思路和实现步骤：

核心解决思路 #

绕过外层固定Schema的限制，把焦点放在解析Message Body中的业务JSON数据上，针对这部分业务数据启用Schema Evolution，再同步到Delta表中。

具体实现步骤 #

1. 使用Autoloader读取外层Avro数据 #

先通过Autoloader读取ADLS上由Event Hub Capture生成的Avro文件，此时得到的DataFrame包含外层固定字段：

df = (spark.readStream
      .format("cloudFiles")
      .option("cloudFiles.format", "avro")
      .load("/path/to/adls/eventhub-capture/"))

此时df的Schema是Event Hub Capture的固定结构，其中Body字段为二进制类型。

2. 解析Message Body为业务JSON并处理Schema演化 #

将二进制的Body转换为字符串，再通过Spark的JSON解析工具处理业务数据。这里提供两种适配Schema演化的方式：

方式一：自动推断业务Schema并兼容变化

先从样本数据中推断初始业务Schema，再在解析时允许字段新增：

from pyspark.sql.functions import col, from_json

# 抽取样本Body字符串，推断初始业务Schema
sample_body = df.selectExpr("CAST(Body AS STRING)").limit(1).collect()[0][0]
business_schema = spark.read.json(spark.sparkContext.parallelize([sample_body])).schema

# 解析Body为业务数据，permissive模式容忍Schema变化
business_df = df.selectExpr(
    "CAST(Body AS STRING) AS body_str",
    "Offset",
    "SequenceNumber",
    "PartitionId",
    "EnqueuedTime",
    "ContentType"
).withColumn("business_data", from_json(col("body_str"), business_schema, options={"mode": "permissive"}))

# 扁平化业务数据（按需选择）
flattened_df = business_df.select("business_data.*", "Offset", "SequenceNumber", "PartitionId", "EnqueuedTime")

方式二：借助Autoloader原生Schema Evolution能力

如果业务JSON的Schema变化频繁，可以先将解析后的业务JSON写入临时存储，再用Autoloader读取并启用原生演化：

# 第一步：将业务JSON字符串写入ADLS临时目录
write_query = (business_df.select("body_str")
               .writeStream
               .format("text")
               .option("path", "/path/to/adls/temp-json/")
               .option("checkpointLocation", "/path/to/adls/checkpoint-temp/")
               .start())

# 第二步：用Autoloader读取临时JSON，启用Schema演化
final_df = (spark.readStream
            .format("cloudFiles")
            .option("cloudFiles.format", "json")
            .option("cloudFiles.schemaLocation", "/path/to/adls/schema-store/")
            .option("cloudFiles.schemaEvolutionMode", "addNewColumns")
            .load("/path/to/adls/temp-json/"))

3. 写入Delta表并开启Schema演化 #

无论用哪种方式解析业务数据，最后写入Delta表时需开启合并Schema配置：

(final_df.writeStream
 .format("delta")
 .option("checkpointLocation", "/path/to/adls/checkpoint-delta/")
 .option("mergeSchema", "true")
 .table("uc_catalog.database.business_table"))

mergeSchema=true会让Delta表自动识别并添加新增的业务字段，实现Schema Evolution。

关键注意点 #

• 外层Event Hub Capture的Schema是固定不变的，读取时无需考虑其演化，只需聚焦业务JSON部分。
• permissive模式会让Spark在解析JSON时忽略字段不匹配的错误，避免因Schema变化导致任务中断。
• 复杂嵌套的业务JSON结构同样适用上述方法，Spark会自动推断嵌套Schema并处理演化。

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