Pipeline. Delta Load JSON Files. Filtering out processed files not working

DebbieE — Tue, 01 Oct 2024 14:33:59 GMT

I am trying to load JSON files from a Data Lake into SQL DW using Pipelines

I have a table LOG_PROCESSED_FILE and a stored Procedure. at the end of the process the SP records the processed File

And the current state is that I have processed 2 files and a new file is waiting to be loaded

1. GetJSONFiles Get Meta data activity File List 'Child Items'

Here is the output

{ "childItems": [ { "name": "A.json", "type": "File" }, { "name": "B.json", "type": "File" }, { "name": "C.json", "type": "File" } ], "effectiveIntegrationRuntime": "AutoResolveIntegrationRuntime (UK South)", "executionDuration": 1, "durationInQueue": { "integrationRuntimeQueue": 10 }, "billingReference": { "activityType": "PipelineActivity", "billableDuration": [ { "meterType": "AzureIR", "duration": 0.016666666666666666, "unit": "Hours" } ] } }

2. Lookup LookupProcessedFiles SELECT FileName FROM [framework].[PROCESSED_FILE_LOG] connected to my data warehouse

Here is the output

{ "count": 2, "value": [ { "FileName": "A.json" }, { "FileName": "B.json" } ], "effectiveIntegrationRuntime": "AutoResolveIntegrationRuntime (UK South)", "billingReference": { "activityType": "PipelineActivity", "billableDuration": [ { "meterType": "AzureIR", "duration": 0.016666666666666666, "unit": "Hours" } ], "totalBillableDuration": [ { "meterType": "AzureIR", "duration": 0.016666666666666666, "unit": "Hours" } ] }, "durationInQueue": { "integrationRuntimeQueue": 0 } }

3. A Filter. FilterProcessedFiles

Items: @activity('GetJsonFiles').output.childItems

Condition :

@if(empty(activity('LookupProcessedFiles').output.value), true, not(contains(activity('LookupProcessedFiles').output.value, item().name)))

This is where its going wrong. On the condition. The output is

{ "ItemsCount": 3, "FilteredItemsCount": 3, "Value": [ { "name": "A.json", "type": "File" }, { "name": "Bjson", "type": "File" }, { "name": "C.json", "type": "File" } ] }

and it should be just C

After this I have a foreach activity containing the copy activity and the stored Procedure activity. These seem to be working fine.

I have researched this condition all day and I havent got any further. Could anyone help me understand what the condition needs to be to get this working. Of if I need to change what I have?

Re: Pipeline. Delta Load JSON Files. Filtering out processed files not working

DebbieE — Tue, 01 Oct 2024 14:58:50 GMT

All sorted. The condition needed a slight change

@IF(empty(activity('LookupProcessedFiles').output.value), true, not(contains(string(activity('LookupProcessedFiles').output.value), item().name)))

Now wrapped in String

Re: Pipeline. Delta Load JSON Files. Filtering out processed files not working

Anonymous — Wed, 02 Oct 2024 02:19:23 GMT

Hi, @DebbieE

Glad you found your problem, and you can mark your approach as a solution to help more people with similar problems find answers. Thanks for your understand.

Best Regards,
Yang
Community Support Team

topic Re: Pipeline. Delta Load JSON Files. Filtering out processed files not working in Data Engineering

Pipeline. Delta Load JSON Files. Filtering out processed files not working

Re: Pipeline. Delta Load JSON Files. Filtering out processed files not working

Re: Pipeline. Delta Load JSON Files. Filtering out processed files not working