使用Kafka驱动读取Azure Event Hub无数据输出的调试方法

1. 确认Event Hub是否有可消费的数据 #

• 登录Azure Portal，进入目标Event Hub实例，查看指标中的“传入消息”，确认有消息流入；
• 手动发送一条测试消息（比如用Azure CLI命令az eventhubs event send或Event Hub Explorer工具），验证消息是否能被正常接收。

2. 验证连接配置的准确性 #

• 检查BOOTSTRAP_SERVERS：确认命名空间名称无误，端口9093是Event Hub Kafka协议的正确端口；
• 校验EH_SASL中的连接字符串：
  • 确保SharedAccessKeyName和SharedAccessKey与Azure Portal中配置的完全一致，无多余符号或拼写错误；
  • 注意username固定为$ConnectionString，不要修改；
  • 可以直接写死连接字符串测试，避免变量转义问题：

    EH_SASL = """kafkashaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="Endpoint=sb://myeventhubns.servicebus.windows.net/;SharedAccessKeyName=MyKeyName;SharedAccessKey=myaccesskey";"""
    

• 确认TOPIC名称与Event Hub实体名称完全匹配（大小写敏感）。

3. 检查权限配置 #

• 确认使用的Shared Access Policy（MyKeyName）拥有**Listen（监听）**权限，在Azure Portal的Event Hub命名空间/实体的“共享访问策略”中查看；
• 用该连接字符串通过第三方工具（如kafka-console-consumer.sh）测试消费，排除权限问题：

  kafka-console-consumer.sh --bootstrap-server myeventhubns.servicebus.windows.net:9093 --topic myeventhub --consumer.config consumer.properties
  

  其中consumer.properties包含：

  sasl.mechanism=PLAIN
  security.protocol=SASL_SSL
  sasl.jaas.config=kafkashaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="Endpoint=sb://myeventhubns.servicebus.windows.net/;SharedAccessKeyName=MyKeyName;SharedAccessKey=myaccesskey";
  

4. 查看Spark流作业的状态与日志 #

• 在Databricks的作业页面找到对应流作业，查看运行状态是否正常，有无停滞或报错；
• 查看驱动端与Executor端日志：在作业详情的“日志”标签中，搜索kafka、error、timeout等关键词，排查认证失败、连接超时等问题；
• 用代码验证流DataFrame是否正确创建：

  print(df.isStreaming)  # 应返回True
  

5. 调整流输出配置，增加调试信息 #

• 为写流添加触发间隔与不截断输出的配置，确保能及时看到结果：

  df_write = df.writeStream \
      .outputMode("append") \
      .format("console") \
      .trigger(processingTime='5 seconds') \
      .option("truncate", "false") \
      .start() \
      .awaitTermination()
  

• 尝试用静态读方式测试，排除流处理逻辑问题：

  df_static = spark.read \
      .format("kafka") \
      .option("subscribe", TOPIC) \
      .option("kafka.bootstrap.servers", BOOTSTRAP_SERVERS) \
      .option("kafka.sasl.mechanism", "PLAIN") \
      .option("kafka.security.protocol", "SASL_SSL") \
      .option("kafka.sasl.jaas.config", EH_SASL) \
      .option("kafka.group.id", "test-debug-group") \
      .option("startingOffsets", "earliest") \
      .load()
  df_static.show()
  

6. 检查Databricks集群的网络与兼容性 #

• 验证集群与Event Hub的网络连通性：在集群节点上执行telnet myeventhubns.servicebus.windows.net 9093，确认端口可访问；若集群在VNet内，需配置服务端点或专用链接；
• 确认Spark Kafka连接器版本与Event Hub兼容：Spark 3.x需搭配Kafka连接器2.4及以上版本，可在Databricks集群的“库”页面查看连接器版本。

内容的提问来源于stack exchange，提问作者Mathias Rönnlund