Building an ETL Pipeline with ADF Step by Step
A first ADF pipeline copies data from A to B. A production ETL pipeline must also survive reruns, updated records and broken extracts. This step-by-step guide builds the stage, validate and merge design, shows on real runs why plain appends go wrong, and gives the ADF pipeline and stored procedure.
GloryTecks Team
Training & Placement Faculty
Last updated: 10 October 2026
Building an ETL pipeline with ADF usually starts with the Copy activity: source dataset, sink dataset, run. That is the right first step, and the first-pipeline guide walks through it. But the default sink behaviour is to append, as a bulk insert (Microsoft Learn: Azure SQL Database connector). Appending is correct only when every extract contains new records, arrives once, and is never broken. Production extracts break all three assumptions.
This guide builds the design that handles them: copy into a staging table, validate the batch, merge it into the target in one transaction, and record the outcome. First, the reason it is worth the extra steps.
Why append-only goes wrong
The code below loads the same four daily batches into two designs. Design A appends, like the default Copy sink. Design B stages, validates and merges. The batches are realistic: day 2 updates two existing orders and adds one; day 2 is then rerun, as happens when a later activity fails and someone reruns the pipeline; day 3's extract is broken, with a duplicated order and a missing amount.
import sqlite3
db = sqlite3.connect(":memory:")
db.executescript("""
CREATE TABLE target_append (order_id INTEGER, status TEXT, amount REAL); -- design A: append only
CREATE TABLE staging (order_id INTEGER, status TEXT, amount REAL); -- design B: stage ...
CREATE TABLE target (order_id INTEGER PRIMARY KEY, status TEXT, amount REAL, batch TEXT); -- ... then merge
CREATE TABLE load_audit (batch TEXT, outcome TEXT, rows INTEGER, detail TEXT);
""")
def design_a(rows): # ADF's default sink behaviour is insert
db.executemany("INSERT INTO target_append VALUES (?, ?, ?)", rows)
def design_b(batch, rows):
db.execute("DELETE FROM staging") # the copy's pre-copy script empties staging
db.executemany("INSERT INTO staging VALUES (?, ?, ?)", rows)
problems = db.execute("""
SELECT (SELECT COUNT(*) FROM staging) = 0,
(SELECT COUNT(*) FROM staging WHERE order_id IS NULL OR amount IS NULL),
(SELECT COUNT(*) - COUNT(DISTINCT order_id) FROM staging),
(SELECT COUNT(*) FROM staging WHERE amount < 0)""").fetchone()
if any(problems):
detail = "empty={} missing={} duplicate_keys={} negative={}".format(*problems)
db.execute("INSERT INTO load_audit VALUES (?, 'rejected', ?, ?)", (batch, len(rows), detail))
return
with db: # merge and audit commit together, or not at all
db.execute("""INSERT INTO target SELECT order_id, status, amount, ? FROM staging WHERE true
ON CONFLICT(order_id) DO UPDATE SET status = excluded.status,
amount = excluded.amount, batch = excluded.batch""", (batch,))
db.execute("INSERT INTO load_audit VALUES (?, 'loaded', ?, '')", (batch, len(rows)))
def report(step):
a = db.execute("SELECT COUNT(*), COALESCE(SUM(amount), 0) FROM target_append").fetchone()
b = db.execute("SELECT COUNT(*), COALESCE(SUM(amount), 0) FROM target").fetchone()
print(f"{step:<38} A append: {a[0]:>2} rows, total {a[1]:>7.2f} | B stage+merge: {b[0]:>2} rows, total {b[1]:>7.2f}")
day1 = [(101, "placed", 250.0), (102, "placed", 99.5), (103, "placed", 410.0), (104, "placed", 75.0)]
day2 = [(102, "shipped", 99.5), (103, "cancelled", 0.0), (105, "placed", 120.0)] # two changes, one new
day3 = [(106, "placed", 60.0), (106, "placed", 60.0), (107, "placed", None)] # a broken extract
for step, batch, rows in [("1 day 1 extract", "d1", day1), ("2 day 2 extract", "d2", day2),
("3 day 2 rerun after a later failure", "d2", day2), ("4 day 3 broken extract", "d3", day3)]:
design_a(rows); design_b(batch, rows); report(step)
print("\nB's audit table")
for row in db.execute("SELECT * FROM load_audit"):
print(" ", row)
print("\nB's target after all runs")
for row in db.execute("SELECT * FROM target ORDER BY order_id"):
print(" ", row)1 day 1 extract A append: 4 rows, total 834.50 | B stage+merge: 4 rows, total 834.50
2 day 2 extract A append: 7 rows, total 1054.00 | B stage+merge: 5 rows, total 544.50
3 day 2 rerun after a later failure A append: 10 rows, total 1273.50 | B stage+merge: 5 rows, total 544.50
4 day 3 broken extract A append: 13 rows, total 1393.50 | B stage+merge: 5 rows, total 544.50
B's audit table
('d1', 'loaded', 4, '')
('d2', 'loaded', 3, '')
('d2', 'loaded', 3, '')
('d3', 'rejected', 3, 'empty=0 missing=1 duplicate_keys=1 negative=0')
B's target after all runs
(101, 'placed', 250.0, 'd1')
(102, 'shipped', 99.5, 'd2')
(103, 'cancelled', 0.0, 'd2')
(104, 'placed', 75.0, 'd1')
(105, 'placed', 120.0, 'd2')After four runs, design A holds 13 rows and a total of 1,393.50. The real business state is five orders worth 544.50. Every failure mode contributed: the updated orders were added as new rows instead of replacing the old ones, the rerun loaded the whole day again, and the broken extract went straight in. Design B ended with exactly five orders, the latest status of each, the right total, and an audit row explaining why day 3 was refused.
The design, step by step
Step 1: choose the write behaviour deliberately
The Azure SQL sink supports Insert, the default, and Upsert, which updates rows whose keys exist and inserts the rest; the keys default to the table's primary key. For extra processing before the final insert, the documentation describes loading to a staging table and then invoking a Stored Procedure activity. Use plain upsert when the extract can be trusted; use staging when you need a validation gate.
Step 2: give each target a staging table
The staging table has the same columns as the extract, with no constraints that could make the copy fail halfway. It holds one batch at a time and is emptied before each copy.
Step 3: copy into staging with a pre-copy script
The sink's preCopyScript runs once per copy run, before data is written, so TRUNCATE TABLE stg.orders there guarantees staging holds only this batch. If the copy fails, nothing has touched the target yet.
Step 4: validate the batch
Before the target is touched, check the batch: is it empty, are there missing keys or required values, duplicate keys, impossible values? Day 3 failed two checks. Reject the whole batch, not the bad rows silently, and make the pipeline fail so someone looks. Choose checks from real past incidents rather than inventing dozens.
Step 5: merge in one transaction
Update rows whose key exists, insert the rest, and write the audit row, all in one transaction. Either the whole batch lands or none of it does. Because a merge sets each row to the batch's values, applying the same batch twice gives the same result; the rerun of day 2 changed nothing in design B. That property, idempotency, is what makes it safe to press "rerun".
Step 6: record the outcome
A load audit table, with batch, outcome, row count and reason, answers the questions people ask the morning after: did yesterday load, how many rows, why was it refused?
The ADF pipeline
In Data Factory this is two activities: a Copy activity into staging, then a Stored Procedure activity, of type SqlServerStoredProcedure, that validates and merges (Microsoft Learn: Stored Procedure activity). The batch identifier is a pipeline parameter, typically set from the trigger's date or window.
{
"name": "pl_orders_daily",
"properties": {
"parameters": { "batch": { "type": "string" } },
"activities": [
{
"name": "CopyToStaging",
"type": "Copy",
"inputs": [ { "referenceName": "ds_source_orders", "type": "DatasetReference" } ],
"outputs": [ { "referenceName": "ds_stg_orders", "type": "DatasetReference" } ],
"typeProperties": {
"source": { "type": "DelimitedTextSource" },
"sink": { "type": "AzureSqlSink", "preCopyScript": "TRUNCATE TABLE stg.orders" }
}
},
{
"name": "ValidateAndMerge",
"type": "SqlServerStoredProcedure",
"dependsOn": [ { "activity": "CopyToStaging", "dependencyConditions": [ "Succeeded" ] } ],
"linkedServiceName": { "referenceName": "ls_azure_sql", "type": "LinkedServiceReference" },
"typeProperties": {
"storedProcedureName": "etl.usp_validate_and_merge_orders",
"storedProcedureParameters": {
"batch": { "value": { "value": "@pipeline().parameters.batch", "type": "Expression" }, "type": "String" }
}
}
}
]
}
}The stored procedure carries the logic from the demo. Rejection writes an audit row and raises an error, so the activity and the pipeline fail visibly; the merge and its audit row commit together.
CREATE OR ALTER PROCEDURE etl.usp_validate_and_merge_orders @batch varchar(20)
AS
BEGIN
SET NOCOUNT ON;
SET XACT_ABORT ON;
DECLARE @rows int = (SELECT COUNT(*) FROM stg.orders);
IF @rows = 0
OR EXISTS (SELECT 1 FROM stg.orders WHERE order_id IS NULL OR amount IS NULL OR amount < 0)
OR EXISTS (SELECT order_id FROM stg.orders GROUP BY order_id HAVING COUNT(*) > 1)
BEGIN
INSERT INTO etl.load_audit (batch, outcome, row_count, logged_at)
VALUES (@batch, 'rejected', @rows, SYSUTCDATETIME());
THROW 50001, 'Staging validation failed; target not changed.', 1;
END;
BEGIN TRANSACTION;
MERGE dbo.orders AS t
USING stg.orders AS s ON t.order_id = s.order_id
WHEN MATCHED THEN
UPDATE SET t.status = s.status, t.amount = s.amount, t.batch = @batch
WHEN NOT MATCHED BY TARGET THEN
INSERT (order_id, status, amount, batch) VALUES (s.order_id, s.status, s.amount, @batch);
INSERT INTO etl.load_audit (batch, outcome, row_count, logged_at)
VALUES (@batch, 'loaded', @rows, SYSUTCDATETIME());
COMMIT;
END;Rehearse the failures before go-live
The four batches in the demo make a useful acceptance checklist for the real pipeline. In a development factory, run a normal day, then the same day again, then a day with changed records, then a deliberately broken file. After each run, compare the target's row count and a key total with what you expect, and read the audit table. If the rerun changes anything, or the broken file reaches the target, the design has a gap. Keep the broken file in the repository so the check can be repeated after every change to the procedure.
Where this fits with the other patterns
- Which rows to extract is a separate decision: watermarks or change tracking, covered in incremental loading in ADF.
- Keeping history instead of overwriting the latest values needs a Type 2 dimension; see slowly changing dimensions in ADF.
- What happens after a failure, alerts, failure paths and retries, is covered in error handling in ADF.
The merge here assumes a source that sends changed and new rows. Deletes are not handled: if the source can delete records, either receive a deletion flag in the extract or compare full snapshots periodically.
What is the default write behaviour of the ADF Copy activity into Azure SQL Database?
Why load into a staging table first?
How do I make an ADF pipeline safe to rerun?
Should I use Copy activity upsert or a stored procedure?
Production ETL design with Data Factory and SQL is part of the Data Engineering course.
Sources
GloryTecks Team
Training & Placement Faculty
The GloryTecks faculty team mentors learners across Data Science, AI and Data Engineering, with a focus on real-time projects and placement readiness in Hyderabad.
Want to master Azure Data Factory?
Join GloryTecks’ expert-led Data Engineering training in Hyderabad with real-time projects and 100% placement support.
Azure Data Lake Storage Gen2 Explained
NextADF Control Flow: ForEach, If and Until Activities
Related articles
ADF Pipelines Explained: Building Your First Pipeline
A beginner's guide to pipelines in Azure Data Factory: what a pipeline and an activity are, the three kinds of activity, the objects you create for a first copy from blob storage to a SQL table, the JSON behind the canvas, why Debug moves real data, what Publish does, and how to read a run's output.
Triggers in ADF: Schedule, Tumbling Window and Event
The four Azure Data Factory trigger types — schedule, tumbling window, storage event and custom event — compared on what they guarantee, with a decision table, trigger JSON you can adapt, and the settings that cause missed or duplicate runs.
Mapping Data Flows in ADF: Transformations Without Code
Mapping data flows let you design transformations on a canvas in Azure Data Factory and run them on Spark clusters the service manages. This guide explains how they work, walks through one flow stage by stage with real row counts, and sets out when a data flow is the right tool.