What is a Data Pipeline? Anatomy and Design Principles

SOURCE TYPE          EXAMPLES                    EXTRACTION APPROACH
Relational DB        PostgreSQL, MySQL, Oracle   CDC (Debezium) or SQL incremental
Document DB          MongoDB, Firestore          Change Streams or scheduled export
REST API             Stripe, Salesforce          HTTP pagination with cursor
Event Stream         Kafka, Kinesis, Pub/Sub     Kafka Consumer Group (streaming)
File Drop            SFTP, S3 landing zone       File event trigger or scheduled scan
Webhook              Payment events, IoT         HTTP endpoint + Kafka/DB write

WHAT TO UNDERSTAND ABOUT EACH SOURCE:
  Schema, cardinality, change rate, latency need, quality, access, and
  history — before designing any pipeline, know all seven for its source.

SELECT * FROM orders;   -- every row, every time

# Use for: small tables (<1M rows), reference/dimension tables,
#          tables with no reliable "changed at" timestamp
# Avoid for: large transaction tables, high-velocity sources,
#            sources with rate limits or shared connection pools

import json
from pathlib import Path
from datetime import datetime, timezone

CHECKPOINT = Path('/data/checkpoints/orders.json')

def load_checkpoint() -> datetime:
    if CHECKPOINT.exists():
        return datetime.fromisoformat(json.loads(CHECKPOINT.read_text())['last_updated_at'])
    return datetime(2020, 1, 1, tzinfo=timezone.utc)

def save_checkpoint(ts: datetime) -> None:
    CHECKPOINT.write_text(json.dumps({'last_updated_at': ts.isoformat()}))

last_run, current_run = load_checkpoint(), datetime.now(timezone.utc)
rows = db.query("SELECT * FROM orders WHERE updated_at > %s AND updated_at <= %s",
                 (last_run, current_run))

write_to_destination(rows)
save_checkpoint(current_run)   # advance ONLY after a successful write

# four pitfalls this pattern has to account for:
1. Late-arriving data past the window       → overlap by 30 min, upsert at destination
2. Deletes are invisible to incremental SQL → use CDC (Module 24)
3. Clock skew between pipeline and source   → use the source DB's own NOW()
4. No updated_at column at all              → use max(id) watermark, or full extract

TYPE              EXAMPLE
Type casting      "380.00" → DECIMAL
Null handling     COALESCE(amount, 0)
Deduplication     ROW_NUMBER() OVER (PARTITION BY id)
Filtering         WHERE status != 'test'
Normalisation     LOWER(status), TRIM(name)
Enrichment        JOIN to customers table
Aggregation       SUM, COUNT, AVG, PERCENTILE
Anonymisation     SHA256(email)
Window calc       SUM OVER (PARTITION BY ... ORDER BY ...)

WHERE IT HAPPENS: Python/Pandas (general-purpose, easy to test) ·
SQL/dbt (set-based, best for tabular data) · Spark (distributed, complex) ·
Flink/Spark Streaming (real-time)

TRUNCATE TABLE silver.store_master;
INSERT INTO silver.store_master SELECT * FROM source_store_master;

-- Fix the empty-window risk with a staging table + atomic rename swap:
CREATE TABLE silver.store_master_staging AS SELECT * FROM source_store_master;
ALTER TABLE silver.store_master RENAME TO store_master_old;
ALTER TABLE silver.store_master_staging RENAME TO store_master;
DROP TABLE silver.store_master_old;

INSERT INTO silver.events (event_id, user_id, event_type, ts)
SELECT event_id, user_id, event_type, ts FROM staging.events
WHERE ts > (SELECT MAX(ts) FROM silver.events);
-- add a UNIQUE constraint on event_id + ON CONFLICT DO NOTHING to survive reruns

INSERT INTO silver.orders (order_id, status, amount, updated_at)
VALUES (%s, %s, %s, %s)
ON CONFLICT (order_id) DO UPDATE SET
    status = EXCLUDED.status, amount = EXCLUDED.amount, updated_at = EXCLUDED.updated_at
WHERE silver.orders.updated_at < EXCLUDED.updated_at;
-- the WHERE clause stops a replayed OLD record from overwriting a newer one

-- Snowflake MERGE — same idea:
MERGE INTO silver.orders AS target USING staging.orders AS source
ON target.order_id = source.order_id
WHEN MATCHED AND target.updated_at < source.updated_at THEN UPDATE SET status = source.status
WHEN NOT MATCHED THEN INSERT (order_id, status, amount, updated_at)
    VALUES (source.order_id, source.status, source.amount, source.updated_at);

-- a replayed record with an OLDER updated_at than what's already in the table:
UPDATE 0
-- the WHERE clause silently skipped it — exactly the intended, idempotent behavior

LINEAR — one input, one output, sequential stages
  [PostgreSQL orders] → [Python cleaner] → [S3 Bronze Parquet]
  Simple failure model. No parallelism between stages.

FAN-OUT — one source, multiple sinks (may partially fail)
                        ┌→ [S3 data lake (Parquet)]
  [Kafka payments] ─────┤→ [PostgreSQL (OLTP write-through)]
                        └→ [Elasticsearch (search index)]
  Must decide: fail all if any fail, or allow partial success?

FAN-IN — multiple sources merged into one sink
  [Stripe payments] ─┐
  [Square payments]  ─┤→ [UNION ALL] → [silver.all_payments]
  [Venmo payments]   ─┘
  Must dedup after union — same transaction ID from multiple sources?

DAG — stages with dependencies, some run in parallel (no cycles)
  [Extract orders] ─────┬──────────────────┐
  [Extract customers] ──┤→ [Silver orders] →┤→ [Gold daily revenue]
  [Extract restaurants] ─┘                  └→ [Gold customer LTV]
  A failed upstream stage blocks all downstream stages — this is what Airflow models.

STREAMING — continuous, event-driven, no concept of "a run"
  [Kafka: orders] → [Flink/Spark Streaming] → [Kafka: enriched_orders]
                                             → [Cassandra (real-time store)]
  Failure means falling behind (consumer lag), not stopping completely.

LAMBDA — batch path for accuracy + streaming path for low latency
  [Source] ──┬─ [Batch, nightly] ─────→ [Batch layer (accurate)]
             └─ [Streaming, real-time] → [Speed layer (fast)] → [Serving: merge both]
  Two codebases for the same logic — Kappa (streaming-only) is the modern alternative.

def etl_orders(source_conn, dest_conn):
    raw = pd.read_sql("SELECT * FROM orders WHERE updated_at > %s", source_conn)
    raw = raw.drop_duplicates(subset=['order_id'])
    raw = raw[raw['amount'] > 0]
    raw['status'] = raw['status'].str.lower().str.strip()
    raw['customer_city'] = raw['customer'].apply(lambda x: x.get('city'))  # flatten JSON
    raw.to_sql('silver_orders', dest_conn, if_exists='append', index=False)

# Step 1: EL — load raw data as-is
def extract_load_orders(source_conn, warehouse_conn):
    raw = pd.read_sql("SELECT * FROM orders WHERE updated_at > %s", source_conn)
    raw.to_sql('raw_orders', warehouse_conn, if_exists='append')

# Step 2: models/silver/orders.sql — dbt transforms the raw table in place
# SELECT order_id, amount::DECIMAL(10,2), LOWER(TRIM(status)) AS status
# FROM {{ source('raw', 'orders') }}
# WHERE amount > 0 AND LOWER(status) IN ('placed','confirmed','delivered','cancelled')
# QUALIFY ROW_NUMBER() OVER (PARTITION BY order_id ORDER BY updated_at DESC) = 1

# 2026 default: EL raw into Bronze, dbt transforms Silver/Gold.
# Python ETL only for PII masking, ML features, and complex flattening.

CREATE TABLE monitoring.pipeline_runs (
    run_id UUID PRIMARY KEY, pipeline_name VARCHAR(100) NOT NULL, run_date DATE NOT NULL,
    started_at TIMESTAMPTZ NOT NULL, finished_at TIMESTAMPTZ,
    status VARCHAR(20) NOT NULL,   -- 'running', 'success', 'failed', 'partial'
    rows_extracted BIGINT, rows_written BIGINT, rows_rejected BIGINT,
    duration_seconds DECIMAL(10,2), error_message TEXT, dlq_count INTEGER DEFAULT 0
);

-- day-over-day row count check, run after every load:
SELECT run_date, rows_written, LAG(rows_written) OVER (ORDER BY run_date) prev_day_rows,
    ABS(rows_written - LAG(rows_written) OVER (ORDER BY run_date))
    / NULLIF(LAG(rows_written) OVER (ORDER BY run_date), 0) pct_change
FROM monitoring.pipeline_runs WHERE pipeline_name = 'orders_ingestion'
ORDER BY run_date DESC LIMIT 30;

ALERT CONDITIONS:
status = 'failed'                     → immediate alert
duration_seconds > expected * 2       → SLA warning
rows_written < expected * 0.8         → data quality alert
rows_rejected > total_rows * 0.05     → data quality alert
No row inserted for today by 8 AM    → pipeline did not run at all

"""
Daily orders ingestion: PostgreSQL source → S3 Bronze Parquet
Schedule: 00:30 UTC daily, previous day. Owner: data-team@freshcart.com
Idempotent: yes (upserts on order_id). Resumable: yes (checkpoint per file).
"""
import os, json, logging, uuid
from datetime import date, datetime, timedelta, timezone
from pathlib import Path
from typing import Iterator
import psycopg2
import pyarrow as pa
import pyarrow.parquet as pq

BATCH_SIZE     = 100_000
CHECKPOINT_DIR = Path('/data/checkpoints')
DLQ_DIR        = Path('/data/dlq')

class Config:
    db_url:  str = os.environ['SOURCE_DB_URL']
    s3_path: str = os.environ['S3_OUTPUT_PATH']

def extract_orders(conn, run_date: date) -> Iterator[dict]:
    """Extract all orders for run_date. Fixed window — idempotent for the same date."""
    start_ts = datetime(run_date.year, run_date.month, run_date.day, tzinfo=timezone.utc)
    end_ts   = start_ts + timedelta(days=1)
    with conn.cursor('orders_cursor') as cur:   # server-side cursor: streams rows
        cur.execute("SELECT * FROM orders WHERE created_at >= %s AND created_at < %s",
                     (start_ts, end_ts))
        for row in cur:
            yield dict(zip([d[0] for d in cur.description], row))

def validate_row(row: dict) -> tuple[dict | None, str | None]:
    """Pure function — no I/O, fully unit-testable."""
    if not row.get('order_id'):
        return None, 'missing_order_id'
    if (row.get('amount') or 0) <= 0:
        return None, f'invalid_amount: {row.get("amount")}'
    if row.get('status') not in ('placed', 'confirmed', 'delivered', 'cancelled'):
        return None, f'invalid_status: {row.get("status")}'
    return row, None

def write_parquet_batch(rows: list[dict], path: str) -> None:
    pq.write_table(pa.Table.from_pylist(rows), path, compression='zstd')

def run(run_date: date) -> dict:
    run_id, log = str(uuid.uuid4()), logging.getLogger('orders_ingestion')
    stats = {'run_id': run_id, 'rows_extracted': 0, 'rows_written': 0, 'rows_rejected': 0}
    log.info('Pipeline started', extra={'run_date': str(run_date), 'run_id': run_id})

    conn, batch, chunk = psycopg2.connect(Config.db_url), [], 0
    try:
        for row in extract_orders(conn, run_date):
            stats['rows_extracted'] += 1
            clean, error = validate_row(row)
            if error:
                stats['rows_rejected'] += 1
                with open(DLQ_DIR / f'orders_{run_date}_{run_id}.ndjson', 'a') as f:
                    f.write(json.dumps({'error': error, 'row': row}) + '\n')
                continue
            batch.append(clean)
            if len(batch) >= BATCH_SIZE:
                chunk += 1
                write_parquet_batch(batch, f'{Config.s3_path}/date={run_date}/part-{chunk:05d}.parquet')
                stats['rows_written'] += len(batch)
                batch = []
        if batch:
            chunk += 1
            write_parquet_batch(batch, f'{Config.s3_path}/date={run_date}/part-{chunk:05d}.parquet')
            stats['rows_written'] += len(batch)
    finally:
        conn.close()

    log.info('Pipeline complete', extra=stats)
    return stats

if __name__ == '__main__':
    import sys
    logging.basicConfig(level=logging.INFO, format='%(message)s')
    run_date = date.fromisoformat(sys.argv[1]) if len(sys.argv) > 1 else date.today() - timedelta(days=1)
    result = run(run_date)
    sys.exit(0 if result['rows_rejected'] / max(result['rows_extracted'], 1) < 0.05 else 1)

{"run_date": "2026-03-17", "run_id": "a1f9-...", "msg": "Pipeline started"}
...
{"rows_extracted": 812400, "rows_written": 811980, "rows_rejected": 420, "msg": "Pipeline complete"}
$ echo $?
0   # 420/812400 = 0.05% rejection — well under the 5% exit-code threshold

You are asked to audit the existing orders pipeline and identify what is fragile about it. Here is the original pipeline code you inherit:

# ORIGINAL PIPELINE (from a junior engineer two years ago)
import psycopg2
import pandas as pd

conn = psycopg2.connect("postgresql://admin:password123@prod-db-01:5432/orders")

df = pd.read_sql("SELECT * FROM orders", conn)        # PROBLEM 1
df['amount'] = df['amount'].astype(float)              # PROBLEM 2
df = df.dropna()                                       # PROBLEM 3
df.to_sql('silver_orders', warehouse_conn, if_exists='replace')  # PROBLEM 4
print("done")                                          # PROBLEM 5

Problem 1 — Full extraction every run: reads all 180 million rows every morning, taking 4 hours and slowing production. No incremental pattern (violates Source Isolation).

Problem 2 — Silent type casting failure: astype(float) crashes the entire pipeline the moment one vendor sends a non-numeric amount, which happens weekly (violates Data Quality Enforcement).

Problem 3 — Silent data deletion: dropna() drops every row with any null — orders missing a promo_code (the majority) vanish, and revenue metrics are quietly wrong.

Problem 4 — Truncate-and-replace every run: if_exists='replace' drops and recreates the table every run — empty for the whole 4-hour window (violates Idempotency and Atomicity).

Problem 5 — No observability: the only output is “done” — no row counts, no timing, no run ID to debug with.

After applying the eight design principles, the pipeline becomes the structured, resumable, observable version shown in Part 08. It processes only yesterday’s new orders (incremental), validates each row and routes failures to a DLQ (data quality enforcement), writes in batches with upserts (idempotency), logs structured metrics (observability), and takes 4 minutes instead of 4 hours (source isolation). Every principle has a direct, measurable impact.