如何从Databricks连接并加载数据至Dataverse实体表

1. 建立Databricks与Dataverse的连接步骤 #

• 准备Dataverse环境信息：获取Dataverse的环境URL（格式如https://<环境名称>.crm.dynamics.com）、目标实体表的逻辑名称（可在Dataverse实体设计器中查看）。
• 在Databricks Notebook中安装依赖：执行%pip install pyodata azure-identity，安装用于Dataverse API交互和Azure认证的库。
• 配置Azure AD服务主体：在Azure AD中创建服务主体，为其分配Dataverse环境的系统管理员或对应实体表的写入权限，记录服务主体的client_id、client_secret、tenant_id。

2. 连接方式与认证方式 #

连接方式 #

采用Dataverse Web API进行直接交互，这是官方推荐的编程式连接方案，支持完整的CRUD操作。

认证方式 #

优先使用Azure AD服务主体认证，适配自动化的Notebook任务：

• 所需认证参数：租户ID（tenant_id）、服务主体客户端ID（client_id）、服务主体客户端密钥（client_secret）。
• 临时测试可使用用户交互式认证（如Device Code Flow），但不推荐用于生产环境。

3. Databricks Notebook写入Dataverse的代码示例 #

以下是Python Notebook的完整实现示例：

# 导入依赖库
from azure.identity import ClientSecretCredential
import requests

# 配置核心参数
DATAVERSE_ENV_URL = "https://your-env-name.crm.dynamics.com"
TENANT_ID = "your-tenant-id"
CLIENT_ID = "your-client-id"
CLIENT_SECRET = "your-client-secret"
ENTITY_LOGICAL_NAME = "your-entity-logical-name"  # 示例：account、contact

# 获取Dataverse API访问令牌
credential = ClientSecretCredential(tenant_id=TENANT_ID, client_id=CLIENT_ID, client_secret=CLIENT_SECRET)
token = credential.get_token("https://crm.dynamics.com/.default")
access_token = token.token

# 初始化请求会话
session = requests.Session()
session.headers.update({
    'Authorization': f'Bearer {access_token}',
    'OData-MaxVersion': '4.0',
    'OData-Version': '4.0',
    'Accept': 'application/json',
    'Content-Type': 'application/json; charset=utf-8'
})

# 构造待写入的数据（以account实体为例）
record_data = {
    "name": "测试企业",
    "address1_city": "上海",
    "telephone1": "13900139000",
    "revenue": 1000000
}

# 发送写入请求
response = session.post(
    f"{DATAVERSE_ENV_URL}/api/data/v9.2/{ENTITY_LOGICAL_NAME}s",
    json=record_data
)

# 校验写入结果
if response.status_code == 201:
    print("数据写入成功！")
    print("新建记录ID:", response.json()[f"{ENTITY_LOGICAL_NAME}id"])
else:
    print(f"写入失败，状态码：{response.status_code}")
    print("错误详情:", response.text)

关键注意事项 #

• 确保服务主体拥有Dataverse目标实体表的创建权限，可在Dataverse安全角色中配置细分权限。
• 实体字段必须使用逻辑名称（而非显示名称），可通过Dataverse实体元数据页面查询。
• 批量写入时建议使用$batch请求，减少API调用次数，提升写入效率。

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