data-pipeline

When to Use

Apply when building, reviewing, or debugging data pipelines, ETL jobs, or data orchestration workflows.

Core Rules

• All pipelines must be idempotent — running twice gives the same result as running once
• Prefer ELT over ETL — load raw first, transform in the warehouse (cheaper to re-transform than re-extract)
• Always use incremental loads over full loads for tables > 100K rows
• Fail fast and loudly — a silent partial load is worse than a visible failure
• Test transformations with fixed fixtures, not prod data samples

Idempotency Patterns

# ❌ Non-idempotent — double run = double rows
def load_events(conn, events):
    for event in events:
        conn.execute("INSERT INTO events VALUES (?)", event)

# ✓ Idempotent — delete-then-insert by partition
def load_events(conn, events, date: str):
    conn.execute("DELETE FROM events WHERE date = ?", date)
    conn.executemany("INSERT INTO events VALUES (?)", events)

# ✓ UPSERT — idempotent by natural key
def upsert_users(conn, users):
    conn.executemany("""
        INSERT INTO users (id, email, name, updated_at)
        VALUES (?, ?, ?, ?)
        ON CONFLICT (id) DO UPDATE SET
          email = excluded.email,
          name = excluded.name,
          updated_at = excluded.updated_at
    """, users)

Incremental Load Pattern

from datetime import datetime, timedelta

def get_last_watermark(conn, table: str) -> datetime:
    row = conn.execute(
        "SELECT MAX(watermark) FROM pipeline_state WHERE table_name = ?", table
    ).fetchone()
    return row[0] or datetime(2020, 1, 1)

def update_watermark(conn, table: str, watermark: datetime):
    conn.execute("""
        INSERT INTO pipeline_state (table_name, watermark)
        VALUES (?, ?)
        ON CONFLICT (table_name) DO UPDATE SET watermark = excluded.watermark
    """, (table, watermark))

def load_incremental(source_conn, dest_conn, table: str):
    last = get_last_watermark(dest_conn, table)
    new_watermark = datetime.utcnow()

    rows = source_conn.execute(
        f"SELECT * FROM {table} WHERE updated_at > ?", last
    ).fetchall()

    upsert_rows(dest_conn, table, rows)
    update_watermark(dest_conn, table, new_watermark)