如何在ADF中实现Lookup活动以检查上次触发器运行后新增的行

核心思路 #

利用ADF计划触发器自带的系统变量作为时间过滤基准，动态生成Lookup活动的SQL查询，精准筛选出上次触发器运行后新增的数据行。

具体实现步骤 #

1. 使用触发器系统变量获取基准时间
   计划触发器每次运行时，ADF会自动生成两个关键时间变量：

   • @trigger().scheduledTime：触发器的计划触发时间（推荐用这个，避免实际启动延迟导致漏查）
   • @trigger().startTime：触发器实际启动的时间
     直接在Lookup活动中引用这些变量即可获取上次运行的时间节点。

2. 配置Lookup活动的动态SQL查询

   • 在Lookup活动的「源」设置中，选择你的SQL数据库链接，切换到「查询」模式。
   • 输入动态SQL语句，用表达式注入触发器时间变量：

     SELECT COUNT(*) AS NewRowCount FROM input_table WHERE audit_datetime > '@{trigger().scheduledTime}'
     

   • 如果数据库的audit_datetime格式与变量默认格式不匹配，用ADF日期函数转换格式：

     SELECT COUNT(*) AS NewRowCount FROM input_table 
     WHERE audit_datetime > CONVERT(datetime, '@{formatDateTime(trigger().scheduledTime, 'yyyy-MM-dd HH:mm:ss')}', 120)
     

3. 处理边界情况（可选）
   如果担心触发器中断导致漏查，可将上次成功运行时间持久化存储：

   • 在SQL数据库中创建状态表adf_run_history，包含last_successful_run_time列。
   • 管道开头先通过Lookup活动获取该表的时间值，用此值作为过滤条件；管道运行成功后，再用存储过程或Copy活动更新该时间为当前触发器的scheduledTime。

4. 利用Lookup结果执行后续逻辑
   Lookup活动的输出会包含查询得到的NewRowCount，可通过If Condition活动判断数值：

   • 当NewRowCount > 0时，触发后续数据处理分支
   • 当数值为0时，直接跳过后续步骤

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