Azure Data Factory v2：自定义Execute Pipeline activity配置及ETL流程问询

我来分享一套针对你这个场景的落地方案，亲测在多个ETL项目中都能很好解决动态触发、重复取数和错误处理的问题：

1. 核心逻辑梳理 #

你的需求本质是做一个调度控制层：Master Pipeline作为入口，先从数据库拉取配置规则，判断对应Child Pipeline是否需要执行，再通过参数动态指定要运行的子管道；同时还要解决重复取数和全局错误兜底的问题。

2. 具体落地步骤 #

2.1 先搭好数据库配置表 #

首先得有个配置表来存规则，比如我习惯建pipeline_execution_config，字段可以这么设：

• child_pipeline_name：子管道的唯一名称（和你要传的参数完全对应）
• is_enabled：是否允许执行（1=启用，0=禁用）
• last_success_run_time：子管道上次成功跑完的时间（核心用来做重复取数的过滤）
• source_data_check_col：可选，比如源表的更新时间字段，用来判断是否有新数据

2.2 Master Pipeline的组件配置 #

第一步：拉取配置信息

用Lookup活动查询配置表，过滤出当前要处理的子管道：

SELECT is_enabled, last_success_run_time 
FROM pipeline_execution_config 
WHERE child_pipeline_name = '@{pipeline().parameters.child_pipeline_name}'

这样就能拿到当前子管道的启用状态和上次成功时间。

第二步：判断是否要跑子管道

加个If Condition活动，判断逻辑可以根据你的需求调整，比如我常用的：

@and(
    equals(activity('Lookup_Config').output.firstRow.is_enabled, 1),
    // 判断源数据是否有更新，或者直接判断上次运行时间是否超过阈值
    greater(utcnow(), addminutes(activity('Lookup_Config').output.firstRow.last_success_run_time, 1440))
)

简单说就是：子管道已启用，且距离上次成功运行已经超过一天（或者你自定义的周期），才触发执行。

第三步：动态指定子管道

在Execute Pipeline活动里，把「Pipeline name」直接绑定到参数：
@pipeline().parameters.child_pipeline_name
这样不管你传哪个子管道名，都能精准触发。

2.3 彻底解决重复取数问题 #

• 在Child Pipeline读取源数据时，直接用配置表里的last_success_run_time做过滤，比如SQL查询：

  SELECT * FROM source_table 
  WHERE update_time > '@{pipeline().parameters.last_success_run_time}'
  

  这里的last_success_run_time可以从Master Pipeline通过Execute Pipeline的参数传递过来。
• 当Child Pipeline执行成功后，Master Pipeline用Stored Procedure活动更新配置表的last_success_run_time为当前时间：

  UPDATE pipeline_execution_config 
  SET last_success_run_time = GETUTCDATE()
  WHERE child_pipeline_name = '@{pipeline().parameters.child_pipeline_name}'
  

3. ETL错误处理的兜底方案 #

3.1 子管道错误的即时告警 #

在Execute Pipeline活动的Failure分支里，加个邮件或Webhook活动，把错误信息发出来，比如：

子管道「@{pipeline().parameters.child_pipeline_name}」执行失败！
错误详情：@{activity('Execute_Child_Pipeline').error.message}
时间：@{utcnow()}
同时可以把错误信息写入专门的etl_error_log表，方便后续排查。

3.2 全局错误捕获 #

在Master Pipeline的根节点添加Failure触发器，捕获所有未被分支处理的错误，统一记录日志并告警，避免出现“管道跑崩了没人知道”的情况。
另外，对Lookup、存储过程这些关键步骤，记得在活动的「Settings」里设置重试次数和间隔，比如失败了重试2次，每次间隔30秒，减少偶发错误的影响。

4. 一些优化小技巧 #

• 把配置查询的逻辑封装成存储过程，Lookup活动直接调用存储过程，后期修改规则更方便。
• 如果子管道需要更多参数（比如源表名、过滤条件），可以在配置表加对应的字段，Master Pipeline读取后传递给子管道。
• 用管道运行历史+自定义仪表盘做监控，随时能看到每个子管道的执行状态、数据同步量这些指标。

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