Spark Streaming如何仅读取Azure Event Hub指定分区？

你的问题出在**setStartingPositions仅用于设置指定分区的起始偏移位置，并不会限制Spark消费的分区范围**，所以默认情况下依然会订阅所有5个分区。要实现仅读取分区"0"和"4"，并让检查点仅跟踪这两个分区的偏移，需要通过EventHubsConf的setPartitionIds方法明确指定要消费的分区列表。

修正后的代码示例 #

val name = "my_event_hub"
val connectionString = "my_event_hub_connection_string"
val max_events = 50

// 指定要消费的分区ID（部分版本库要求传入String类型，如"0"、"4"）
val targetPartitions = Seq("0", "4")

val positions = Map(
  new NameAndPartition(name, 0) -> EventPosition.fromEndOfStream,
  new NameAndPartition(name, 4) -> EventPosition.fromEndOfStream
)

val eventHubsConf = EventHubsConf(connectionString)
                    .setPartitionIds(targetPartitions) // 关键：指定仅消费这些分区
                    .setStartingPositions(positions) // 修正变量名：原代码中start应为positions
                    .setMaxEventsPerTrigger(max_events) // 修正变量名：原代码中max_Events应为max_events

关键说明 #

1. setPartitionIds的作用：该方法会让Spark结构化流仅连接并订阅你指定的分区，不会处理其他未列出的分区，检查点也只会维护这些分区的偏移信息，完全匹配你的需求。
2. 变量名修正：原代码存在两处变量名错误，会导致代码无法正常运行，需同步修正。
3. 版本适配：确保使用的azure-event-hubs-spark库版本支持setPartitionIds方法（建议使用2.3.15及以上版本），这是官方推荐的标准分区过滤方式。

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