无连接字符串时用Service Principal将Event Hub数据读入Spark DataFrame

前置要求 #

• 你的Spark集群已安装和Spark/Scala版本匹配的azure-eventhubs-spark连接器，例如Spark 3.3+版本建议使用2.3.21及以上版本的azure-eventhubs-spark_2.12
• 已获取Service Principal的三个核心参数：客户端ID(Client ID)、客户端密钥(Client Secret)、租户ID(Tenant ID)
• 已为该Service Principal分配目标Event Hub（或其父级命名空间）的Azure Event Hubs Data Receiver角色权限

核心实现（PySpark示例） #

from pyspark.sql import SparkSession

# 初始化SparkSession
spark = SparkSession.builder.appName("EventHubStreamRead").getOrCreate()

# 替换为你的实际参数值
EVENTHUB_NAMESPACE = "<你的Event Hub命名空间>"
EVENTHUB_NAME = "<你的Event Hub名称>"
CONSUMER_GROUP = "<你的消费者组，默认值为$Default>"
SP_CLIENT_ID = "<Service Principal客户端ID>"
SP_CLIENT_SECRET = "<Service Principal客户端密钥>"
SP_TENANT_ID = "<Azure租户ID>"

# 构造Event Hub连接配置
eh_config = {
    "eventhubs.namespace": EVENTHUB_NAMESPACE,
    "eventhubs.name": EVENTHUB_NAME,
    "eventhubs.consumerGroup": CONSUMER_GROUP,
    # 仅声明OAuth认证，无需填写共享访问密钥
    "eventhubs.connection.string": f"Endpoint=sb://{EVENTHUB_NAMESPACE}.servicebus.windows.net/;EntityPath={EVENTHUB_NAME};Authentication=OAuth",
    "eventhubs.oauth.client.id": SP_CLIENT_ID,
    "eventhubs.oauth.client.secret": SP_CLIENT_SECRET,
    "eventhubs.oauth.tenant.id": SP_TENANT_ID,
    "eventhubs.oauth.endpoint": f"https://login.microsoftonline.com/{SP_TENANT_ID}/oauth2/token"
}

# 读取流数据生成Spark DataFrame
stream_df = spark.readStream.format("eventhubs").options(**eh_config).load()

# 查看默认返回的DataFrame结构
stream_df.printSchema()

# 测试示例：将数据输出到控制台
# query = stream_df.writeStream.format("console").start()
# query.awaitTermination()

注意事项 #

• 默认读取的DataFrame中body字段为二进制格式，可根据你的数据编码格式（如JSON、CSV）调用对应的解析函数转换为结构化数据
• Service Principal权限分配后最长可能需要15分钟生效，如出现权限报错可先等待后重试
• 如使用VPC或私有端点访问Event Hub，需要确认Spark集群的网络策略允许访问Event Hub端点

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