在Delta Live Tables中加载解析AVRO文件失败的技术求助

问题背景 #

之前在普通Databricks环境中，可通过RDD转换解析AVRO文件中的JSON Body字段，但在Delta Live Tables（DLT）流水线中复现该流程时，解析后的视图无法识别字段名称。已通过Auto Loader将AVRO文件加载到Bronze层（以Delta格式存储原始AVRO数据），尝试创建视图解析JSON字段但失败。

原普通Spark可行代码 #

in_path = '/mnt/file_location/*/*/*/*/*.avro'
avroDf = spark.read.format("com.databricks.spark.avro").load(in_path)
jsonRdd = avroDf.select(avroDf.Body.cast("string")).rdd.map(lambda x: x[0])
data = spark.read.json(jsonRdd)

data.createOrReplaceTempView("eventhub")

# 查询数据
sql_query1 = sqlContext.sql("""
select distinct
data.field.test1             as col1
,data.field.test2             as col2
,data.field.fieldgrp.city     as city
from
    eventhub
""")

DLT中失败的代码 #

@dlt.view(name=f"eventdata_v")
def eventdata_v():
    avroDf = spark.read.format("delta").table("live.bronze_file_list")
    jsonRdd = avroDf.select(avroDf.Body.cast("string")).rdd.map(lambda x: x[0])
    data = spark.read.json(jsonRdd)
    return data

@dlt.view(name=f"eventdata2_v")
def eventdata2_v():
    df = (
        dlt.read("eventdata_v")
        .select("data.field.test1 ")
    )
    return df

问题根源 #

1. DLT的声明式执行模式对Schema的处理逻辑与普通Spark不同，spark.read.json(jsonRdd)的自动Schema推断在DLT流水线中无法正确注册元数据，导致后续视图无法识别字段。
2. RDD转换破坏了DataFrame的优化链路，DLT无法对RDD操作进行增量优化，也难以稳定捕获Schema信息。

正确实现方式 #

使用Spark内置的from_json函数直接解析JSON字符串字段，结合显式Schema定义确保字段被正确识别，完全避免RDD转换：

步骤1：定义JSON字段的Schema #

根据实际JSON结构明确定义Schema，避免自动推断的不确定性：

from pyspark.sql.types import StructType, StructField, StringType, IntegerType
from pyspark.sql.functions import col, from_json

# 按实际JSON结构调整字段名称与数据类型
json_schema = StructType([
    StructField("data", StructType([
        StructField("field", StructType([
            StructField("test1", StringType(), nullable=True),
            StructField("test2", IntegerType(), nullable=True),
            StructField("fieldgrp", StructType([
                StructField("city", StringType(), nullable=True)
            ]))
        ]))
    ]))
])

步骤2：在DLT视图中解析JSON #

用from_json解析Bronze层的Body字段，直接生成结构化DataFrame：

@dlt.view(name="eventdata_v")
def eventdata_v():
    # 使用dlt.read读取DLT内部表，符合流水线元数据管理规范
    bronze_df = dlt.read("bronze_file_list")
    # 解析JSON并展开顶层结构
    return bronze_df.select(
        from_json(col("Body").cast("string"), json_schema).alias("parsed_data")
    ).select("parsed_data.*")

步骤3：创建查询视图 #

现在可以正常引用解析后的层级字段：

@dlt.view(name="eventdata2_v")
def eventdata2_v():
    return dlt.read("eventdata_v").select(
        "data.field.test1",
        "data.field.test2",
        "data.field.fieldgrp.city"
    ).distinct()

额外优化建议 #

• 如果JSON Schema可能发生变化，可开启DLT的Schema演进功能，配合from_json的mode="PERMISSIVE"参数处理兼容的Schema变更。
• 始终使用dlt.read而非spark.read.format("delta").table()读取DLT内部表，DLT会自动处理增量逻辑和元数据同步。

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