Azure Databricks：工作流内触发另一个工作流并复用ADF运行参数

原生支持的实现方式 #

Databricks工作流原生支持跨Job触发并传递参数，步骤如下：

• 在当前工作流中添加一个Job类型的任务，选择需要触发的目标工作流（已创建好的Job）。
• 在任务的参数配置面板中，直接引用当前工作流从ADF获取的run_parameters变量。例如，若ADF传入的参数包含source_table和execution_date，则在目标Job的参数中填写：

  {"source_table": "{{run.parameters.source_table}}", "execution_date": "{{run.parameters.execution_date}}"}
  

• 目标工作流需提前配置对应的参数占位符（在Job的任务设置中定义参数），确保能接收传递的值。

自定义Notebook实现方式（适用于需要额外逻辑的场景） #

如果需要在触发前做参数校验、转换等额外操作，可以通过Notebook调用Databricks Jobs API实现：

• 创建一个Notebook任务，编写以下Python代码：

  import requests
  import json
  
  # 获取当前工作流的run_parameters
  current_params_str = dbutils.widgets.get("run_parameters")
  current_params = json.loads(current_params_str)
  
  # 获取Databricks环境信息
  api_url = dbutils.notebook.entry_point.getDbutils().notebook().getContext().apiUrl().get()
  api_token = dbutils.notebook.entry_point.getDbutils().notebook().getContext().apiToken().get()
  
  # 触发目标工作流
  target_job_id = "替换为目标工作流的Job ID"
  request_payload = {
      "job_id": target_job_id,
      "parameters": current_params
  }
  
  response = requests.post(
      f"{api_url}/api/2.1/jobs/run-now",
      headers={"Authorization": f"Bearer {api_token}"},
      json=request_payload
  )
  
  if response.status_code == 200:
      print("目标工作流触发成功")
  else:
      raise Exception(f"触发失败: {response.text}")
  

• 确保执行工作流的身份（服务主体或用户）拥有目标Job的jobs.runNow权限。

总结 #

优先使用原生Job任务触发方式，配置简单且无需额外代码；若需要灵活的参数处理或触发逻辑，再选择Notebook调用API的方案。

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