如何从Azure Synapse触发Databricks Delta Live Tables流水线？

方法1：使用Databricks REST API #

直接调用DLT流水线的运行API触发执行，步骤如下：

• 获取必要参数：

  • 工作区URL：格式为https://<databricks-instance>.azuredatabricks.net
  • API令牌：从Databricks工作区的用户设置中生成
  • DLT流水线ID：从流水线页面的URL中提取（/pipelines/后的字符串）

• Notebook中执行Python代码触发流水线：

import requests
import json

# 替换为你的实际参数
databricks_url = "https://<你的工作区URL>"
api_token = "<你的API令牌>"
pipeline_id = "<你的DLT流水线ID>"

# 构建API请求
endpoint = f"{databricks_url}/api/2.0/pipelines/{pipeline_id}/updates"
headers = {
    "Authorization": f"Bearer {api_token}",
    "Content-Type": "application/json"
}
# full_refresh设为True表示全量刷新，False为增量刷新
request_body = {"full_refresh": False}

# 发送请求并处理响应
response = requests.post(endpoint, headers=headers, json=request_body)
if response.status_code == 200:
    print("DLT流水线触发成功！")
    print(f"运行ID: {response.json()['update_id']}")
else:
    print(f"触发失败：{response.text}")

方法2：使用Databricks Python SDK（官方推荐） #

通过官方SDK简化调用流程，无需手动处理HTTP请求：

• 先在Notebook中安装SDK（若未安装）：

%pip install databricks-sdk

• 编写代码触发流水线：

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.pipelines import FullRefresh

# 自动使用当前Notebook的上下文认证
w = WorkspaceClient()

# 触发流水线，替换为你的流水线ID
update = w.pipelines.start_update(
    pipeline_id="<你的DLT流水线ID>",
    full_refresh=FullRefresh.FALSE  # 全量刷新设为FullRefresh.TRUE
)

print(f"DLT流水线已触发，运行ID: {update.update_id}")

关键注意事项 #

• 运行Notebook的用户需拥有DLT流水线的「可运行（Can Run）」权限
• 若要从Azure Synapse触发，只需在Synapse流水线中调用包含上述代码的Databricks Notebook，确保权限配置正确

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