Forum Discussion

DebbieE's avatar
DebbieE
Community Champion
1 year ago
Solved

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

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?
 
  • 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

2 Replies

  • DebbieE's avatar
    DebbieE
    Community Champion

    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

  • Anonymous's avatar
    Anonymous
    Not applicable

    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