如何在Azure Databricks中使用PySpark消费Event Hub事件？

没问题！我来给你梳理下在Azure Databricks里用PySpark消费Event Hub事件的完整步骤——其实和Scala版本的核心逻辑一致，但配置和代码细节有一些区别，直接上干货：

1. 先搞定依赖包 #

Azure Event Hub的Spark连接器基于Scala开发，但PySpark可以直接使用。你需要在Databricks集群中添加对应的Maven依赖：

• 依赖坐标：com.microsoft.azure:azure-eventhubs-spark_2.12:2.3.22（注意：这个版本适配Spark 3.x，如果你用的是旧版Databricks Runtime，要对应调整连接器版本）

添加方式：

• 打开你的Databricks集群 → 进入Libraries标签 → 点击Install New → 选择Maven → 输入上述坐标并安装
• 或者在Notebook开头用魔法命令临时安装：

  %maven com.microsoft.azure:azure-eventhubs-spark_2.12:2.3.22
  

2. 配置Event Hub连接参数 #

你需要准备Event Hub的核心连接信息，用字典存储这些参数，有两种配置方式：

方式一：直接使用完整连接字符串 #

从Azure Portal的Event Hub实例中复制包含EntityPath的连接字符串（注意要选对SAS政策，必须有Listen权限）：

eh_config = {
  "eventhubs.connectionString": "Endpoint=sb://<你的命名空间>.servicebus.windows.net/;SharedAccessKeyName=<SAS政策名称>;SharedAccessKey=<SAS密钥>;EntityPath=<Event Hub名称>",
  "eventhubs.consumerGroup": "$Default"  # 可选，默认就是这个消费者组，也可以自定义
}

方式二：拆分参数配置 #

如果不想暴露完整连接字符串，也可以拆分配置：

eh_config = {
  "eventhubs.namespaceName": "<你的命名空间>",
  "eventhubs.eventHubName": "<Event Hub名称>",
  "eventhubs.sasKeyName": "<SAS政策名称>",
  "eventhubs.sasKey": "<SAS密钥>",
  "eventhubs.consumerGroup": "$Default"
}

3. 读取Event Hub事件（两种模式） #

模式一：流式读取（实时消费） #

适合处理实时产生的事件，Spark Streaming会持续监听Event Hub：

from pyspark.sql.functions import col, binary_to_string

# 初始化流式DataFrame
stream_df = spark.readStream \
  .format("eventhubs") \
  .options(**eh_config) \
  .load()

# 解析Event Hub的二进制消息体
# Event Hub的body字段是二进制类型，需要转成字符串再处理（如果是JSON消息可以进一步解析）
parsed_stream = stream_df \
  .select(
    binary_to_string(col("body")).alias("message_content"),
    col("enqueuedTime").alias("event_time"),  # 事件进入Event Hub的时间
    col("partitionId").alias("partition"),
    col("offset").alias("event_offset")
  )

# 输出到控制台（测试用），也可以写入Delta Lake、ADLS等持久化存储
# 注意：流式处理必须设置checkpointLocation来保证容错
query = parsed_stream.writeStream \
  .outputMode("append") \
  .format("console") \
  .option("truncate", False) \
  .option("checkpointLocation", "/dbfs/FileStore/eventhub_checkpoint/")  # 用DBFS路径存检查点
  .start()

# 等待流式任务结束（如果要手动停止，在Notebook里可以用query.stop()）
query.awaitTermination()

模式二：批量读取（一次性消费历史数据） #

适合一次性读取Event Hub中的历史事件：

from pyspark.sql.functions import col, binary_to_string

# 初始化批量DataFrame
batch_df = spark.read \
  .format("eventhubs") \
  .options(**eh_config) \
  .load()

# 解析消息体
parsed_batch = batch_df \
  .select(
    binary_to_string(col("body")).alias("message_content"),
    col("enqueuedTime").alias("event_time")
  )

# 查看结果
parsed_batch.show(truncate=False)

4. 进阶配置技巧 #

• 控制读取起始位置：可以通过eventhubs.startingPosition指定从哪里开始消费：

  # 从最早的事件开始读
  eh_config["eventhubs.startingPosition"] = "earliest"
  # 从最新的事件开始读
  eh_config["eventhubs.startingPosition"] = "latest"
  # 或者指定具体偏移量（JSON格式）
  eh_config["eventhubs.startingPosition"] = '{"offset": "123", "seqNo": 456, "enqueuedTime": "2024-01-01T00:00:00Z", "isInclusive": true}'
  

• 设置消费批次大小：通过eventhubs.maxEventsPerTrigger控制每次触发读取的事件数量，避免压力过大：

  eh_config["eventhubs.maxEventsPerTrigger"] = 1000
  

5. 常见问题排查 #

• 依赖版本不兼容：确保连接器版本和Databricks Runtime的Spark版本匹配（比如Runtime 11.x对应Spark 3.3，连接器用2.3.22+）
• 权限错误：检查SAS政策是否有Listen权限，连接字符串中的密钥是否正确
• 消息解析失败：如果body是JSON格式，转成字符串后可以用from_json函数解析成结构化数据：

  from pyspark.sql.types import StructType, StringType, IntegerType
  
  schema = StructType() \
    .add("id", IntegerType()) \
    .add("content", StringType())
  
  parsed_stream = stream_df \
    .select(from_json(binary_to_string(col("body")), schema).alias("event_data")) \
    .select("event_data.*")
  

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