How to Implement a Lookup Activity in ADF to Check Rows Added After the Last Trigger Run

Question

I have an ADF pipeline that contains a scheduled trigger to run the pipeline every two minutes. The pipeline's input table is linked to an SQL database in SSMS, and it detects any DDL changes. The input table has several columns, including event, request SQL, and audit datetime. Now, I want to implement a lookup activity in ADF that checks the rows added to the input database after the last trigger run. The lookup activity can use an SQL query like 'count * from input table where audit datetime > time of the last trigger run'. How can I achieve this?

Answer 1

You can use three lookup activities for this scenario. One is to check the datetime of last trigger run time. Second one is to check the current date time. Third one is to check the count of records in the log between these two datetime. Below is the detailed approach.

• Input tables log_tab and wm_tab is taken as in below image. wm_tab is used to store the last trigger run datetime. Initially, value is assigned with 1900-01-01 00:00:00.0000000

• In ADF, Lookup1 is taken and query for source dataset is given as select * from wm_tab. This will take and store the value of last triggered date value.

• Lookup2 is taken and query for source dataset is given as select getdate() as current_datetime. This will store the current date value.

• Lookup3 activity is taken, and it is connected to Lookup1 and lookup2 in such a way that it gets executed only after lookup1 and lookup2 gets executed. Query is given as,

select count(1) as count_of_new_records from log_tab where
audit_time >= '@{activity('Lookup1').output.firstRow.audit_time}' and 
audit_time < '@{activity('Lookup2').output.firstRow.current_datetime}'

• script activity is taken to update the current date time value as new watermark value in wm_tab

update wm_tab
set
audit_time='@{activity('Lookup2').output.firstRow.current_datetime}'

Result

• case:1 When pipeline is run for the first time, count is shown as 5 records. Pipeline checks the count of rows from 1900-01-01 to current date time.

• Also wm_tab is updated with new value.

• case:2 I have not inserted any records after the time 2023-03-23 01:57:40.36000000 and when pipeline is run, it shows count of new records as 0