Case study 02 · Financial Services

Azure Data Factory Enterprise Pipeline

A metadata-driven ingestion framework that onboards a new source table in minutes instead of days.

Data Engineer — framework design, pipeline development, monitoring 3 months

Sources onboarded: 60+ tables
Onboarding time: 3 days → 30 min
Success rate: 99.4%

Executive summary

Built a reusable, metadata-driven ingestion framework in Azure Data Factory that lands data from Oracle, SQL Server, REST APIs and SFTP into ADLS Gen2 as Parquet. One set of parameterised pipelines, driven by a SQL control table, replaced dozens of hand-built copy pipelines and introduced consistent logging, watermarking and alerting.

Business problem

Each new data request produced a new hand-built pipeline. After two years the factory contained more than 70 near-identical pipelines with inconsistent naming, no shared logging and no reliable way to tell what had loaded.

Three days of development and testing to onboard each new source table
Failures discovered by report users rather than by the data team
Full reloads of large tables because incremental logic was implemented inconsistently
Credentials stored in linked services instead of a central vault

Architecture design

A parent/child pipeline pattern: a master pipeline reads active sources from the control table and fans out to typed child pipelines per connector, all landing into a date-partitioned raw zone.

L1
Control plane
Azure SQL control and log tables define what to load, how, and record every outcome.
L2
Orchestration
Master pipeline → ForEach (batch count 8) → connector-specific child pipelines.
L3
Landing (ADLS Gen2)
raw/<system>/<table>/load_date=YYYY-MM-DD/ in Parquet with Snappy compression.
L4
Security
Managed identity everywhere; secrets resolved at runtime from Azure Key Vault.
L5
Observability
Log Analytics diagnostics, alert rules and a Python health report delivered every morning.

Source systems

System	Technology	Volume	Cadence
Core banking	Oracle 19c (self-hosted IR)	34 tables · ~40M rows/day	Hourly incremental
Operational DBs	SQL Server	21 tables	Nightly incremental
Market data	REST API (paginated JSON)	~2M records/day	Every 4 hours
Partner files	SFTP (CSV, fixed-width)	15 file types	Event-based

Data flow

01Trigger
Schedule and storage-event triggers start the master pipeline with a run context.
02Lookup
Active sources for the run's priority group are read from etl.IngestionControl.
03Fan-out
ForEach executes the matching child pipeline per source in parallel.
04Copy
Incremental sources query WHERE watermark > LastWatermark; output lands as Parquet.
05Log & advance
A stored procedure records row counts and advances the watermark only on success.
06Alert
Failures route to Teams and email with pipeline, activity and error message.

Data model

The framework's own model is operational: a control table describing sources and a log table capturing every run — the basis for SLA reporting.

Facts

etl.IngestionLog
Grain: One row per source table per pipeline run
etl.PipelineRunSummary
Grain: One row per master pipeline run

Dimensions

etl.IngestionControl
Source, load type, watermark column, target path, priority
etl.SourceSystem
Connection metadata and owning team
etl.LoadPriority
Priority groups mapped to schedules and SLAs

ETL process

Parameterised datasets

One dataset per connector type with schema, table and path parameters — no per-table datasets.

Watermark safety

The new watermark is captured before the copy and only committed after success, so failures never skip data.

Schema drift

Parquet with schema captured per load; a nightly job compares schemas and flags added or removed columns.

Retry policy

Copy activities retry 3× with a 2-minute interval for transient network failures.

Tech stack

Orchestration: Azure Data FactorySelf-hosted IRTriggers
Storage: ADLS Gen2ParquetAzure SQL Database
Security: Managed identityAzure Key VaultRBAC
Monitoring: Log AnalyticsAzure MonitorPython SDK
DevOps: Git integrationAzure DevOps YAMLARM templates

SQL example

Control table and safe watermark advance (T-SQL)

sql

CREATE TABLE etl.IngestionControl (
    SourceId         INT IDENTITY(1,1) PRIMARY KEY,
    SourceSystem     VARCHAR(50)   NOT NULL,
    SourceSchema     VARCHAR(128)  NOT NULL,
    SourceTable      VARCHAR(128)  NOT NULL,
    LoadType         VARCHAR(12)   NOT NULL
                     CHECK (LoadType IN ('FULL', 'INCREMENTAL')),
    WatermarkColumn  VARCHAR(128)  NULL,
    LastWatermark    DATETIME2(3)  NULL,
    TargetPath       VARCHAR(400)  NOT NULL,
    PriorityGroup    TINYINT       NOT NULL DEFAULT 5,
    IsActive         BIT           NOT NULL DEFAULT 1
);
GO

-- Called by the final Stored Procedure activity of each child pipeline
CREATE OR ALTER PROCEDURE etl.usp_CompleteIngestion
    @SourceId       INT,
    @NewWatermark   DATETIME2(3),
    @RowsCopied     BIGINT,
    @PipelineRunId  UNIQUEIDENTIFIER
AS
BEGIN
    SET NOCOUNT ON;
    SET XACT_ABORT ON;

    BEGIN TRANSACTION;

        UPDATE etl.IngestionControl
        SET    LastWatermark = @NewWatermark
        WHERE  SourceId = @SourceId
          AND (LastWatermark IS NULL OR LastWatermark < @NewWatermark);

        INSERT INTO etl.IngestionLog
               (SourceId, PipelineRunId, RowsCopied, Watermark, Status, LoggedAt)
        VALUES (@SourceId, @PipelineRunId, @RowsCopied, @NewWatermark,
                'Succeeded', SYSUTCDATETIME());

    COMMIT TRANSACTION;
END;

Python example

Morning pipeline health report with the ADF SDK

python

"""Lists failed and slow ADF pipeline runs from the last 24 hours."""
import os
from datetime import datetime, timedelta, timezone

from azure.identity import DefaultAzureCredential
from azure.mgmt.datafactory import DataFactoryManagementClient
from azure.mgmt.datafactory.models import RunFilterParameters

RESOURCE_GROUP = os.environ["ADF_RESOURCE_GROUP"]
FACTORY = os.environ["ADF_NAME"]
SLOW_THRESHOLD_MS = 30 * 60 * 1000

client = DataFactoryManagementClient(
    DefaultAzureCredential(), os.environ["AZURE_SUBSCRIPTION_ID"]
)

now = datetime.now(timezone.utc)
window = RunFilterParameters(
    last_updated_after=now - timedelta(hours=24),
    last_updated_before=now,
)

runs = client.pipeline_runs.query_by_factory(RESOURCE_GROUP, FACTORY, window).value
failed = [r for r in runs if r.status == "Failed"]
slow = [
    r for r in runs
    if r.status == "Succeeded" and (r.duration_in_ms or 0) > SLOW_THRESHOLD_MS
]

for run in failed:
    activities = client.activity_runs.query_by_pipeline_run(
        RESOURCE_GROUP, FACTORY, run.run_id, window
    ).value
    step = next((a for a in activities if a.status == "Failed"), None)
    message = (step.error or {}).get("message", "") if step else ""
    print(f"[FAILED] {run.pipeline_name} | activity={step.activity_name if step else 'n/a'}"
          f" | {message[:200]}")

for run in slow:
    print(f"[SLOW]   {run.pipeline_name} | {run.duration_in_ms / 60000:.1f} min")

success_rate = 1 - len(failed) / max(len(runs), 1)
print(f"Runs: {len(runs)} | Success rate: {success_rate:.2%}")

Challenges

Challenge

Oracle extracts through the self-hosted IR saturated the gateway during peak hours.

Solution

Introduced priority groups and capped ForEach concurrency per group; heavy tables moved to an off-peak schedule.

Challenge

A failed copy previously advanced the watermark and silently skipped rows.

Solution

Captured the new watermark before the copy and committed it in the same transaction as the success log entry.

Challenge

The REST API returned inconsistent page sizes and occasional HTTP 429s.

Solution

Used ADF pagination rules with an end condition and a retry policy with back-off.

Performance improvements

Metric	Before	After	Change
New source onboarding	~3 days	~30 min	−98%
Pipelines to maintain	70+	6	−91%
Hourly batch duration	41 min	14 min	−66%
Mean time to detect failure	Next business day	< 5 min	Real-time alerts

Business impact

Data requests delivered in the same sprint instead of queuing for weeks
Every load is auditable: row counts, watermark and duration per table per run
Incidents detected by the data team before report users notice
Credentials removed from pipeline definitions — a finding closed in the security audit

Lessons learned

Invest in the framework once; the second and hundredth source should be configuration, not code.
Watermarks must only advance after a confirmed success — ordering matters.
Monitoring is part of the product. A pipeline without alerting is not finished.

Success metrics

Tables under framework: 60+Across four source types
Pipeline success rate: 99.4%Trailing 12 months
Daily rows ingested: ~45MIncremental, not full reloads
Alert to triage: < 15 minMedian during business hours

Next case studyMicrosoft Fabric Lakehouse Solution