Databricks中EventHub流读取时Schema不匹配未触发badRecordsPath问题咨询

没错，这确实是Databricks里EventHub ReadStream的默认行为，我来帮你拆解下原因和正确的处理方式：

首先得搞清楚EventHub Spark连接器的默认读取逻辑：它返回的数据流是一个包含元数据的结构体，核心字段包括body（二进制类型，存着你的业务事件内容）、partition、offset、enqueuedTime这些。你直接在readStream里指定schema=badSchema时，Spark只会验证这个顶层结构体是否和你给的schema匹配——只要你的badSchema里包含了这些顶层字段（或者连接器能自动兼容），读取操作就不会报错，自然也不会触发badRecordsPath的逻辑，哪怕body里的内容完全不符合你预期的业务结构。

你的临时方案（对比inferSchema和自定义Schema）是可行的，但其实有更规范的方式来触发badRecordsPath的功能：把事件读取和业务数据验证拆成两步来做：

1. 先读取原始的EventHub流，不需要指定业务schema
2. 单独解析body字段（把二进制转成字符串后，用from_json函数），这时候再传入你的业务schema和badRecordsPath配置，这样当body内容和schema不匹配时，就会自动把坏记录写入指定路径

给你个具体的代码示例：

# 第一步：读取原始EventHub流，获取包含元数据的DataFrame
raw_event_stream = (spark.readStream 
 .format("eventhubs") 
 .options(**eventHubsConf) 
 .load() 
)

# 第二步：解析body字段并启用坏记录捕获
from pyspark.sql.functions import from_json, col

# 把二进制的body转成字符串，再用from_json验证schema
parsed_stream = raw_event_stream.withColumn(
    "business_data",
    from_json(
        col("body").cast("string"),
        schema=badSchema,  # 这里传入你的业务schema
        options={"badRecordsPath": "/tmp/badRecordsPath/test1"}
    )
)

这样处理的话，任何不符合badSchema的业务事件都会被捕获到你指定的路径里，同时正常记录会被解析成结构化的business_data字段。

总结下：EventHub连接器本身的schema验证只针对顶层元数据，而业务数据的验证需要你在解析body字段时单独配置，这才是badRecordsPath功能的正确打开方式。

内容的提问来源于stack exchange，提问作者T.UK