Designing Data Pipelines in Azure

Anatomy of a production-grade pipeline

A production data pipeline in Azure typically has these layers:

┌─────────────────────────────────────────────────────────────────┐
│                    Pipeline Orchestration (ADF/Synapse Pipelines)│
│                                                                  │
│  ┌────────────┐  ┌────────────┐  ┌─────────────┐  ┌─────────┐  │
│  │ Get        │  │ Extract    │  │ Transform   │  │ Load &  │  │
│  │ Watermark  │→ │ to Bronze  │→ │ Bronze→Gold │→ │ Update  │  │
│  │ (Lookup)   │  │ (Copy)     │  │ (Notebook)  │  │ WM (SP) │  │
│  └────────────┘  └────────────┘  └─────────────┘  └─────────┘  │
│                       ↓ on failure                               │
│                  ┌────────────────┐                              │
│                  │ Alert + dead   │                              │
│                  │ letter logging │                              │
│                  └────────────────┘                              │
└─────────────────────────────────────────────────────────────────┘

Each step is an activity in ADF or Synapse Pipelines. The watermark control table (a small table in Azure SQL Database or Synapse) records the last successful load timestamp. Each successful run updates it. If the pipeline fails, the watermark is not updated, so the next run reprocesses from the correct point.

The watermark pattern for incremental loading

The watermark pattern is the foundation of reliable incremental loading. It works for any source that has a column tracking when records were last modified (created_at, modified_at, updated_timestamp, etc.).

-- Control table in Azure SQL Database (or Synapse dedicated pool)
CREATE TABLE dbo.pipeline_watermarks
(
    pipeline_name       NVARCHAR(100)   NOT NULL PRIMARY KEY,
    source_table        NVARCHAR(200)   NOT NULL,
    last_load_time      DATETIME2       NOT NULL,
    last_rows_loaded    BIGINT          NOT NULL DEFAULT 0,
    last_run_status     NVARCHAR(20)    NOT NULL DEFAULT 'SUCCESS',
    last_run_at         DATETIME2       NOT NULL DEFAULT GETDATE()
);

-- Initial seed
INSERT INTO dbo.pipeline_watermarks (pipeline_name, source_table, last_load_time)
VALUES
    ('OrderLoad',    'dbo.Orders',    '2020-01-01 00:00:00'),
    ('CustomerLoad', 'dbo.Customers', '2020-01-01 00:00:00');

-- Stored procedure called at the END of a successful pipeline run
CREATE PROCEDURE dbo.UpdateWatermark
    @pipeline_name  NVARCHAR(100),
    @new_watermark  DATETIME2,
    @rows_loaded    BIGINT
AS
BEGIN
    UPDATE dbo.pipeline_watermarks
    SET
        last_load_time    = @new_watermark,
        last_rows_loaded  = @rows_loaded,
        last_run_status   = 'SUCCESS',
        last_run_at       = GETDATE()
    WHERE
        pipeline_name = @pipeline_name;
END;

# In the ADF Notebook Activity (PySpark):
# Read the watermark from the control table via JDBC

import pyodbc
from datetime import datetime

def get_watermark(pipeline_name: str, connection_string: str) -> datetime:
    conn = pyodbc.connect(connection_string)
    cursor = conn.cursor()
    cursor.execute(
        "SELECT last_load_time FROM dbo.pipeline_watermarks WHERE pipeline_name = ?",
        pipeline_name
    )
    row = cursor.fetchone()
    conn.close()
    return row[0] if row else datetime(2020, 1, 1)

last_load = get_watermark("OrderLoad", jdbc_conn_string)
print(f"Loading orders modified after: {last_load}")

# Extract only records modified since the last watermark
new_orders = spark.read.format("jdbc") \
    .option("url", jdbc_url) \
    .option("dbtable", f"(SELECT * FROM dbo.Orders WHERE modified_date > '{last_load}') AS orders") \
    .load()

rows_loaded = new_orders.count()
print(f"Extracted {rows_loaded} new/changed orders")

Idempotency: safe re-runs

A pipeline is idempotent if running it multiple times with the same inputs produces the same result as running it once. Idempotent pipelines are essential for production systems because any pipeline can fail at any step and must be safely restartable.

The most common idempotency patterns:

# Pattern 1: MERGE INTO (Delta Lake) - idempotent upsert
# Running this twice for the same data produces the same result
from delta.tables import DeltaTable

delta_table = DeltaTable.forPath(spark, silver_path)
delta_table.alias("existing").merge(
    new_orders.alias("incoming"),
    "existing.order_id = incoming.order_id"
).whenMatchedUpdateAll() \
 .whenNotMatchedInsertAll() \
 .execute()

# Pattern 2: OVERWRITE a full partition
# If the pipeline re-runs for the same day, it overwrites that day's data
# Result is always correct regardless of how many times it runs
new_orders \
    .write.format("delta") \
    .mode("overwrite") \
    .option("replaceWhere", "order_date = '2025-03-19'") \
    .save(silver_path)

# Pattern 3: Write to a temp path then atomic rename
# Consumers never see partial data
import uuid
temp_path = f"{silver_path}_temp_{uuid.uuid4().hex[:8]}"
new_orders.write.format("delta").mode("overwrite").save(temp_path)
# Then rename/move atomically - atomicity is guaranteed by Delta's transaction log

Data quality gates

A production pipeline should validate data at the bronze-to-silver transition and fail loudly rather than silently propagating bad data to downstream consumers.

from pyspark.sql import DataFrame
from pyspark.sql.functions import col, isnull, isnan, count, when
import json

def run_quality_checks(df: DataFrame, table_name: str, thresholds: dict) -> bool:
    """
    Run data quality checks and raise an exception if any threshold is breached.
    Returns True if all checks pass.
    """
    total_rows = df.count()
    if total_rows == 0:
        print(f"WARNING: {table_name} has 0 rows — skipping quality checks")
        return True

    quality_results = {}

    # Check null rates for critical columns
    for col_name in thresholds.get("not_null", []):
        null_count = df.filter(isnull(col(col_name))).count()
        null_rate = null_count / total_rows
        quality_results[f"{col_name}_null_rate"] = null_rate
        if null_rate > thresholds.get("max_null_rate", 0.01):
            raise ValueError(
                f"QUALITY FAILURE: {col_name} null rate {null_rate:.1%} "
                f"exceeds threshold {thresholds['max_null_rate']:.1%}"
            )

    # Check for negative values in amount columns
    for col_name in thresholds.get("non_negative", []):
        negative_count = df.filter(col(col_name) < 0).count()
        negative_rate = negative_count / total_rows
        if negative_rate > 0:
            raise ValueError(
                f"QUALITY FAILURE: {col_name} has {negative_count} negative values"
            )

    # Check row count is within expected range
    if "min_rows" in thresholds and total_rows < thresholds["min_rows"]:
        raise ValueError(
            f"QUALITY FAILURE: {table_name} has {total_rows} rows, "
            f"expected at least {thresholds['min_rows']}"
        )

    # Log results to a quality tracking table in Delta
    results_df = spark.createDataFrame([{
        "table_name": table_name,
        "check_timestamp": str(datetime.utcnow()),
        "total_rows": total_rows,
        "results_json": json.dumps(quality_results),
        "passed": True
    }])
    results_df.write.format("delta").mode("append").save(
        "abfss://monitoring@storage.dfs.core.windows.net/data_quality_log/"
    )

    print(f"Quality checks PASSED for {table_name}: {total_rows} rows")
    return True

# Usage
run_quality_checks(
    df=silver_orders,
    table_name="silver.orders",
    thresholds={
        "not_null": ["order_id", "customer_id", "order_date", "order_total"],
        "non_negative": ["order_total"],
        "min_rows": 1000,
        "max_null_rate": 0.001
    }
)

Pipeline monitoring and alerting

A pipeline nobody monitors might as well not exist. Here is the minimal monitoring setup for a production Azure data pipeline:

# Create an Azure Monitor alert for ADF pipeline failures
# This sends an email whenever any ADF pipeline fails

az monitor metrics alert create \
  --name "ADF Pipeline Failure Alert" \
  --resource-group rg-data \
  --scopes "/subscriptions/{sub}/resourceGroups/rg-data/providers/Microsoft.DataFactory/factories/adf-mycompany-prod" \
  --condition "count PipelineFailedRuns > 0" \
  --window-size 5m \
  --evaluation-frequency 5m \
  --action-group "/subscriptions/{sub}/resourceGroups/rg-data/providers/microsoft.insights/actionGroups/DataTeamAlerts" \
  --description "Fires when any ADF pipeline run fails"

# Create an alert for long-running pipelines (possible hangs)
az monitor metrics alert create \
  --name "ADF Pipeline Long Running Alert" \
  --resource-group rg-data \
  --scopes "/subscriptions/{sub}/resourceGroups/rg-data/providers/Microsoft.DataFactory/factories/adf-mycompany-prod" \
  --condition "average PipelineElapsedTime > 7200" \
  --window-size 30m \
  --evaluation-frequency 30m \
  --action-group "/subscriptions/{sub}/resourceGroups/rg-data/providers/microsoft.insights/actionGroups/DataTeamAlerts" \
  --description "Fires when any ADF pipeline runs longer than 2 hours"

Beyond ADF-level alerting, add custom metrics from your Spark notebooks to Azure Monitor using the Application Insights SDK or Log Analytics workspace. Track: rows loaded per run, processing time per stage, data quality check results, and lag (how far behind the pipeline is from the source).

Handling schema evolution

Sources add columns, rename columns, and change data types. A fragile pipeline breaks when this happens. A resilient pipeline adapts:

# Read raw data with schema merge enabled (Delta feature)
# If the incoming data has new columns, Delta adds them to the table schema
new_data = spark.read.json(bronze_path)

new_data.write.format("delta") \
    .mode("append") \
    .option("mergeSchema", "true") \
    .save(silver_path)
# With mergeSchema=true, new columns in new_data are added to the Delta table
# Existing rows have NULL for the new columns

# For type changes (incompatible evolution), you need explicit handling
from pyspark.sql.functions import col, try_cast

# If 'order_total' changed from STRING to DECIMAL in the source:
new_data_fixed = new_data.withColumn(
    "order_total",
    try_cast(col("order_total"), "decimal(18,2)")
)
# try_cast returns NULL for values that cannot be converted (instead of erroring)
# Your quality checks will then catch unexpectedly high NULL rates