如何在Spark中高效读取同一Event Hubs命名空间下的多个Event Hub？

针对你在Databricks环境中需要合并读取同一命名空间下多个同Schema Event Hub流的需求，提供以下两种可行方案：

方案1：明确指定目标Event Hub列表 #

为每个目标Event Hub生成独立配置，分别读取流后通过union合并为单一数据流（因Schema一致，可直接合并）。

# 基础配置信息
ev_namespace    = "{your_namespace_name}"
ev_sas_key_name = "{your_key_name}"
ev_sas_key_val  = "{your_key_value}"
consumer_group = "eventparket-cg"

# 要读取的Event Hub名称列表
eventhub_names = ["{your_eventhub_name_1}", "{your_eventhub_name_2}", "{your_eventhub_name_7}"]

# 封装生成Event Hub配置的函数
def create_eh_conf(eventhub_name):
    conn_string = f"Endpoint=sb://{ev_namespace}.servicebus.windows.net/;EntityPath={eventhub_name};SharedAccessKeyName={ev_sas_key_name};SharedAccessKey={ev_sas_key_val}"
    ehConf = {}
    ehConf['eventhubs.connectionString'] = sc._jvm.org.apache.spark.eventhubs.EventHubsUtils.encrypt(conn_string)
    ehConf['eventhubs.consumerGroup'] = consumer_group
    return ehConf

# 批量读取各Event Hub流并合并
streams = []
for eh_name in eventhub_names:
    eh_conf = create_eh_conf(eh_name)
    stream = spark.readStream.format("eventhubs").options(**eh_conf).load()
    streams.append(stream)

# 合并所有流
combined_df = streams[0]
for stream in streams[1:]:
    combined_df = combined_df.union(stream)

方案2：自动发现命名空间下所有Event Hub #

通过Azure管理API自动获取命名空间内全部Event Hub，再批量读取合并。需确保Databricks集群拥有Microsoft.EventHub/namespaces/eventhubs/read权限。

from azure.mgmt.eventhub import EventHubManagementClient
from azure.identity import DefaultAzureCredential

# Azure资源标识信息
subscription_id = "{your_subscription_id}"
resource_group = "{your_resource_group}"
ev_namespace = "{your_namespace_name}"

# 初始化Event Hub管理客户端
credential = DefaultAzureCredential()
eh_client = EventHubManagementClient(credential, subscription_id)

# 获取命名空间下所有Event Hub名称
eventhub_list = eh_client.event_hubs.list_by_namespace(resource_group, ev_namespace)
eventhub_names = [eh.name for eh in eventhub_list]

# 后续读取合并步骤同方案1
ev_sas_key_name = "{your_key_name}"
ev_sas_key_val = "{your_key_value}"
consumer_group = "eventparket-cg"

def create_eh_conf(eventhub_name):
    conn_string = f"Endpoint=sb://{ev_namespace}.servicebus.windows.net/;EntityPath={eventhub_name};SharedAccessKeyName={ev_sas_key_name};SharedAccessKey={ev_sas_key_val}"
    ehConf = {}
    ehConf['eventhubs.connectionString'] = sc._jvm.org.apache.spark.eventhubs.EventHubsUtils.encrypt(conn_string)
    ehConf['eventhubs.consumerGroup'] = consumer_group
    return ehConf

streams = []
for eh_name in eventhub_names:
    eh_conf = create_eh_conf(eh_name)
    stream = spark.readStream.format("eventhubs").options(**eh_conf).load()
    streams.append(stream)

combined_df = streams[0]
for stream in streams[1:]:
    combined_df = combined_df.union(stream)

注意事项 #

• 需保证所有目标Event Hub的Schema完全一致，否则union操作会报错；若存在细微差异，需先统一Schema再合并。
• 消费组可共用同一组，也可根据业务需求为每个Event Hub单独指定。
• 自动发现方案需安装依赖包，在Databricks中可通过%pip install azure-mgmt-eventhub azure-identity完成安装。

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