Eventhub Stream 未捕获模式不匹配

Question

当我们从 eventthub 中读取事件时，我们正在尝试实现 badRecordsPath，例如，为了尝试使其正常工作，我已经放入了应该使事件失败的架构：

eventStreamDF = (spark.readStream
  .format("eventhubs")
  .options(**eventHubsConf)
  .option("badRecordsPath", "/tmp/badRecordsPath/test1")
  .schema(badSchema)
  .load()
) 

然而这永远不会失败并且总是读取事件，这是 databricks 的 eventthub 的读取流的行为吗？目前的解决方法是根据我们自己的架构检查 inferSchema。

Answer 1

EventHubs 中的数据架构是固定的（请参阅docs）（对于 Kafka 也是如此） - 实际的有效负载始终编码为名称为 body 的二进制字段，开发人员负责解码此二进制有效负载根据数据的生产者和该数据的消费者之间的“联系”。因此，即使您指定了架构和 badRecordsPath 选项，它们也不会被使用。

您将需要实现一些函数来解码来自 JSON 或其他内容的数据，例如，如果数据损坏则返回 null，然后您将有一个空值过滤器将流拆分为两个子流 -好的和坏的数据。