无法从Azure Event Hubs向Databricks流式传输数据：查询无结果

问题概述 #

已完成Azure Event Hubs账户配置，获取连接字符串后，Azure门户能看到数据流入及吞吐量数据，但Databricks无论是通过流式读取还是非流式读取，查询均返回空结果。

用到的代码示例 #

流式读取代码 #

eventHubConf = {}
connectionString = """Endpoint=sb://*********.servicebus.windows.net/;SharedAccessKeyName=**********;SharedAccessKey=***********;EntityPath=**********"""

eventHubConf['eventhubs.connectionString'] = sc._jvm.org.apache.spark.eventhubs.EventHubsUtils.encrypt(connectionString)

df = spark.readStream.format("eventhubs") \
    .options(**eventHubConf) \
    .option("eventhubs.consumerGroup", "$Default") \
    .load()

display(df)

非流式读取代码 #

df = spark.read.format("eventhubs").option("eventhubs.connectionString", connectionString).load()

display(df)

排查方向及解决建议 #

• 消费者组偏移量问题：$Default消费者组若已消费过现有数据，Event Hubs会从上次偏移量开始读取，不会返回已消费数据。建议创建新的消费者组，替换代码中eventhubs.consumerGroup参数后重试。
• 连接字符串权限不足：检查SharedAccessKey对应的权限，必须拥有Listen权限。若使用仅含Send权限的密钥，将无法读取数据。
• Event Hub实体路径错误：确认连接字符串中的EntityPath与实际Event Hub名称完全一致，注意大小写敏感。
• Databricks连接器版本不兼容：确保Databricks集群安装了对应Spark版本的Event Hubs连接器。Spark 3.x推荐使用com.microsoft.azure:azure-eventhubs-spark_2.12:2.3.22及以上版本，版本不匹配会导致读取失败。
• 网络访问限制：检查Event Hubs的防火墙规则，确认是否允许Databricks集群的出站IP访问。若Event Hubs设置了IP白名单，需将Databricks的IP添加至白名单。
• 数据格式未转换：Event Hubs中数据默认以二进制存储，display(df)直接展示可能看不到有效内容。可添加selectExpr("CAST(body AS STRING) AS message")转换后再显示。
• 非流式读取的留存期限制：非流式读取默认仅读取Event Hubs当前留存的数据，若数据超出保留期（默认7天）则返回空。可通过配置eventhubs.startingPosition参数指定起始位置，比如{"offset": "-1", "seqNo": -1, "enqueuedTime": null, "isInclusive": true}读取所有留存数据。

内容的提问来源于stack exchange，提问作者Pranav Sai Putta