PySpark通过服务主体认证Event Hub失败，求解决方法

我有一个具备Event Hub主题读取权限的服务主体，尝试用以下PySpark代码读取Event Hub主题：

# Event Hub connection string using Service Principal (SAS)
eventhub_conn_str = f"Endpoint=sb://{namespace}/;SharedAccessKeyName={client_id};SharedAccessKey={client_secret}"

# Setting up the Azure Event Hubs Spark connector
event_hub_config = {
    'eventhubs.connectionString':eventhub_conn_str,
    'eventhubs.consumerGroup': consumer_group,
    'eventhubs.maxEventsPerTrigger': '1000'
}

# Read from Event Hub using Spark Structured Streaming
raw_events = (
    spark.readStream
    .format("eventhubs")
    .option("startingPosition", "@latest")
    .options(**event_hub_config)
    .load()
)

但始终报错：

java.lang.IllegalArgumentException: Input byte array has wrong 4-byte ending unit

我怀疑是认证失败（用任意密码都会出现相同错误），请问如何通过client_id和client_secret完成认证？

你当前的错误根源是误用了SAS连接字符串格式传递服务主体认证信息——服务主体的client_id和client_secret不能直接替换SAS的SharedAccessKeyName和SharedAccessKey，这会导致连接器尝试把client_secret当作SAS密钥做Base64解码，从而抛出字节数组格式错误。

正确的服务主体认证需要通过Azure AD OAuth2实现，具体配置方式如下：

方法1：直接在配置中指定Azure AD参数 #

无需构造SAS连接字符串，直接在event_hub_config中添加认证相关配置项：

event_hub_config = {
    'eventhubs.connectionString': f"Endpoint=sb://{namespace}/;EntityPath={eventhub_name}",
    'eventhubs.consumerGroup': consumer_group,
    'eventhubs.maxEventsPerTrigger': '1000',
    # Azure AD服务主体认证参数
    'eventhubs.authentication': 'AAD',
    'eventhubs.aadClientId': client_id,
    'eventhubs.aadClientSecret': client_secret,
    'eventhubs.aadTenantId': tenant_id  # 你的Azure租户ID
}

注意：这里的connectionString仅需包含Endpoint和EntityPath，不需要任何SAS或密钥信息。

方法2：使用TokenCredential对象（适用于Spark 3.2+及新版Event Hub连接器） #

如果使用较新版本的Spark和Event Hub连接器，可通过Azure Identity库生成TokenCredential传递给连接器：

1. 先安装依赖包：

pip install azure-identity

2. 修改代码：

from azure.identity import ClientSecretCredential

# 创建服务主体凭证
credential = ClientSecretCredential(
    tenant_id=tenant_id,
    client_id=client_id,
    client_secret=client_secret
)

# 获取Event Hub的访问token
token = credential.get_token("https://eventhubs.azure.net/.default")

event_hub_config = {
    'eventhubs.connectionString': f"Endpoint=sb://{namespace}/;EntityPath={eventhub_name}",
    'eventhubs.consumerGroup': consumer_group,
    'eventhubs.maxEventsPerTrigger': '1000',
    'eventhubs.authentication': 'AAD',
    'eventhubs.token': token.token
}

关键注意事项 #

• 确保服务主体已被授予Azure Event Hubs Data Receiver角色（或更高权限的Event Hub数据访问角色），作用范围覆盖目标Event Hub命名空间或主题。
• 确认使用的Event Hub Spark连接器版本支持Azure AD认证（推荐2.3.20及以上版本）。
• 严格区分SAS认证与Azure AD服务主体认证的格式，两者的连接字符串和配置参数完全不同。

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