如何在Azure Data Factory序列管道中定位失败子管道并获取其RunId？

方案一：通过管道变量记录子管道运行状态（推荐，无需外部API） #

步骤： #

1. 在PL_Main管道中创建变量

   • 新建字符串变量 firstFailedRunId（存储首个失败子管道的RunId）
   • （可选）新建数组变量 failedPipelineRunIds（存储所有失败子管道的RunId）

2. 配置每个Execute Pipeline活动的失败分支
   对PL_1、PL_2、PL3.1、PL_3.2每个活动：

   • 打开活动的设置选项卡，将失败操作设置为Continue（确保子管道失败后，PL_Main继续执行后续活动）
   • 添加失败分支，在分支内添加Set Variable活动：
     • 对于firstFailedRunId，使用表达式：

       @if(empty(variables('firstFailedRunId')), activity('<子管道活动名称>').output.runId, variables('firstFailedRunId'))
       

       （仅当变量为空时存入当前失败子管道的RunId，保证记录首个失败的实例）
     • （可选）对于failedPipelineRunIds，使用表达式追加失败RunId：

       @concat(variables('failedPipelineRunIds'), array(activity('<子管道活动名称>').output.runId))
       

3. 在后续活动中引用变量
   PL_Main完成后，后续的New Activity可以直接通过@pipeline().variables.firstFailedRunId获取首个失败子管道的RunId，作为活动参数传入。

方案二：调用ADF管理API查询子管道运行记录 #

步骤： #

1. 配置Web活动调用子运行API
   在PL_Main完成后的活动中添加Web活动：

   • URL：替换占位符为你的ADF信息

     https://management.azure.com/subscriptions/{订阅ID}/resourceGroups/{资源组名}/providers/Microsoft.DataFactory/factories/{数据工厂名}/pipelineruns/{pipeline().RunId}/childruns?api-version=2018-06-01
     

   • 方法：GET
   • 身份验证：选择Managed Identity，资源设置为https://management.azure.com/（需确保数据工厂的MSI拥有Data Factory Contributor或相关权限）

2. 筛选并提取首个失败的RunId

   • 添加Filter活动，输入为@activity('<Web活动名称>').value，筛选条件设为@equals(item().status, 'Failed')
   • 添加First活动，输入为@activity('<Filter活动名称>').output.value，提取首个失败的子运行记录
   • 后续New Activity可以通过@activity('<First活动名称>').output.runId获取目标RunId

方案对比 #

内容的提问来源于stack exchange，提问作者João Miguens