How to perform incremental cleansing of CSV files in Azure Data Factory

This tutorial explains how to implement a pipeline in Azure Data Factory to detect new or modified CSV files, validate/clean the data and load only the incremental records into a table in Azure SQL. It is useful to reduce costs and avoid duplication when processing recurring file ingestions.

Prerequisites

• Azure account with permissions to create resources.
• An Azure Data Factory v2 resource created.
• Blob Storage or ADLS Gen2 with some sample CSV files.
• An Azure SQL database with target table and write access.
• Basic knowledge of pipelines, datasets and activities in Azure Data Factory.

Step 1: Concept — how incremental ingestion works

Simple explanation: we use the file list (Get Metadata) to detect new/changed files based on lastModified. We store a record of the last processing (watermark) and only process files with lastModified later than that. This avoids reprocessing files that were already validated.

Step 2: Create a control file (watermark) in Blob

We will store the datetime of the last processing in a small JSON file in the blob. Create a file named watermark.json with initial content:

{
  "lastRun":"1970-01-01T00:00:00Z"
}

Step 3: Pipeline — get list of new files

In ADF create a pipeline with the following activities: Get Metadata to list files, Lookup to read the watermark.json, Filter to select only files with lastModified > watermark.

// Get Metadata dataset: point to the CSVs folder, field: childItems
// Lookup dataset: point to watermark.json
// Example expression in the Filter activity to compare dates
@greater(item().lastModified, pipeline().parameters.lastWatermark)

Step 4: Read the lastRun from the watermark (Lookup)

Add a Lookup that reads watermark.json. In the Settings pane, enable First row only. Store the value in a pipeline variable called lastWatermark using a Set Variable with the expression:

@activity('LookupWatermark').output.value.lastRun

Step 5: Use Get Metadata to get files' lastModified

Get Metadata with the folder dataset (Field list: Child Items). The output gives the names; for each item you want the lastModified — you can use a Lookup per file or, preferably, enable a parametrized dataset and use a ForEach activity over the childItems to run an individual Get Metadata that returns lastModified.

// Inside the ForEach (items: @activity('GetFolder').output.childItems)
// Get Metadata (param: fileName) -> field: lastModified
// Expose output: activity('GetFileMetadata').output.lastModified

Step 6: Filter new files and perform simple cleansing

Inside the ForEach, after obtaining lastModified, use an If Condition activity to test if lastModified > lastWatermark. If true, run a Data Flow or Copy with mapping and simple validations (e.g., remove rows with required fields empty, normalize dates).

// If Condition expression
@greater(formatDateTime(activity('GetFileMetadata').output.lastModified,'yyyy-MM-ddTHH:mm:ssZ'), variables('lastWatermark'))

// Example of simple transformations in a Mapping Data Flow:
// - Source: CSV
// - Derived Column: trim() and parseDate()
// - Filter: isNotNull(key_field)
// - Sink: Azure SQL (upsert mode with unique key)

Step 7: Incremental load into Azure SQL

To avoid duplicates, use the Sink of the Data Flow with update/insert (upsert) based on a key column (for example, id or combination of fields). Alternative: load into a staging table and run a stored procedure to deduplicate with MERGE.

// Minimal MERGE example (Azure SQL) to deduplicate after staging load
MERGE dbo.Target AS T
USING dbo.Staging AS S
ON T.Key = S.Key
WHEN MATCHED THEN UPDATE SET T.Col = S.Col
WHEN NOT MATCHED THEN INSERT (Key, Col) VALUES (S.Key, S.Col);

Step 8: Update the watermark after success

At the end of the pipeline, after confirming successful load, write to watermark.json the maximum datetime of the processed files (for example, now()). Use a Web activity to call the Blob REST API (or a Copy activity with a JSON output dataset) to overwrite the file.

// Example expression for new watermark
@utcNow()  // or the maximum among processed files

// To write with Copy activity: source = a small table/variable, sink = watermark.json dataset

Verify the result

Validate: 1) the pipeline ran without errors; 2) the target table contains only the expected records; 3) watermark.json was updated with the new date; 4) re-running the pipeline does not reprocess files already processed. Use Azure Data Factory Monitor and queries in Azure SQL to see changes.

Conclusion

With this pattern you can process CSV files incrementally, validate and load into Azure SQL reducing cost and duplication. Next steps: add detailed logging, handle errors with dead-letter (staging) and parameterize for multiple folders. Tip: start by testing with few files and always check the timezone of the dates.