Databricks托管标识下DefaultAzureCredential令牌获取失败求助

问题概述 #

在Databricks中尝试用DefaultAzureCredential向Azure Service Bus发送消息时，触发如下错误：

ServiceBusError: Handler failed: DefaultAzureCredential failed to retrieve a token from the included credentials.

用户日常通过az login完成本地认证，但不了解Databricks环境下的认证适配方式。

错误原因 #

DefaultAzureCredential默认不会自动识别Databricks环境中的用户会话，它会依次尝试环境变量、托管标识、VS Code等认证路径，但在Databricks集群中这些默认路径无法获取有效令牌，导致认证失败。

解决方案 #

方案1：使用Service Bus连接字符串（快速验证） #

直接从Azure Portal获取Service Bus的连接字符串，存入Databricks Secrets后读取，跳过Credential认证流程：

操作步骤：

1. 在Azure Portal的Service Bus命名空间中，复制「RootManageSharedAccessKey」对应的连接字符串
2. 在Databricks中创建Secret Scope，将连接字符串存入（例如Scope命名为servicebus，Key命名为connection_string）
3. 修改代码如下：

import nest_asyncio
import asyncio
from azure.servicebus import ServiceBusMessage
from azure.servicebus.aio import ServiceBusClient

nest_asyncio.apply()

# 从Databricks Secrets读取连接字符串
CONNECTION_STRING = dbutils.secrets.get(scope="servicebus", key="connection_string")
TOPIC_NAME = "xxoutbound"

async def send_single_message(sender):
    message = ServiceBusMessage("Single Message")
    await sender.send_messages(message)
    print("Sent a single message")

async def run():
    # 使用连接字符串初始化Service Bus Client
    async with ServiceBusClient.from_connection_string(
        conn_str=CONNECTION_STRING,
        logging_enable=True) as servicebus_client:
        sender = servicebus_client.get_topic_sender(topic_name=TOPIC_NAME)
        async with sender:
            await send_single_message(sender)

asyncio.run(run())
print("Done sending messages")
print("-----------------------")

方案2：使用Azure托管标识（生产环境推荐） #

为Databricks工作区或集群启用托管标识，给该标识分配Service Bus的「Azure Service Bus Data Sender」角色，再修改Credential配置：

import nest_asyncio
import asyncio
from azure.servicebus import ServiceBusMessage
from azure.servicebus.aio import ServiceBusClient
from azure.identity.aio import DefaultAzureCredential

nest_asyncio.apply()

FULLY_QUALIFIED_NAMESPACE = "xxx.servicebus.windows.net"
TOPIC_NAME = "xxoutbound"

# 若集群绑定多个托管标识，需指定client_id参数
# credential = DefaultAzureCredential(managed_identity_client_id="<你的托管标识Client ID>")
credential = DefaultAzureCredential()

async def send_single_message(sender):
    message = ServiceBusMessage("Single Message")
    await sender.send_messages(message)
    print("Sent a single message")

async def run():
    async with ServiceBusClient(
        fully_qualified_namespace=FULLY_QUALIFIED_NAMESPACE,
        credential=credential,
        logging_enable=True) as servicebus_client:
        sender = servicebus_client.get_topic_sender(topic_name=TOPIC_NAME)
        async with sender:
            await send_single_message(sender)
        await credential.close()

asyncio.run(run())
print("Done sending messages")
print("-----------------------")

注意事项：

• 需在Azure Portal中给托管标识分配Service Bus命名空间的「Azure Service Bus Data Sender」角色，等待权限生效（通常需10-15分钟）
• 若集群启用多个托管标识，必须指定managed_identity_client_id参数

方案3：适配用户会话认证（仅测试用，不推荐生产） #

如果要复用本地az login的令牌，需手动将令牌传入Databricks环境（存在安全风险，仅用于临时测试）：

# 仅测试场景使用，生产环境禁止
import os
from azure.identity.aio import DefaultAzureCredential

# 从本地az login获取令牌后传入（实际需通过安全方式传递）
os.environ["AZURE_ACCESS_TOKEN"] = "<你的az login令牌>"
credential = DefaultAzureCredential()

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