Data Pipeline

Data Pipelines

A pipeline that runs once and works is easy. A pipeline that runs daily for years and never silently loses data is hard. Design for the latter.

> Reference: reference/pipeline-patterns.md - ETL vs ELT, batch vs streaming, data quality checks, backfill strategies, idempotent writes by DB type.

Pipeline Types

| Type | Pattern | Use Case | |------|---------|----------| | ETL | Extract → Transform → Load | Transform data before loading (data warehouse) | | ELT | Extract → Load → Transform | Load raw, transform in warehouse (modern, preferred) | | CDC | Change Data Capture → Stream → Target | Real-time sync between systems | | Batch | Schedule → Read all → Process → Write | Periodic aggregation, reports | | Streaming | Event → Process → Output (continuous) | Real-time analytics, event processing |

Core Principles

1. Idempotency

Running the pipeline twice with the same input produces the same result. No duplicates, no gaps.

BAD:  INSERT INTO target SELECT * FROM source WHERE date = today
      → Rerun = duplicates

GOOD: DELETE FROM target WHERE date = today;
      INSERT INTO target SELECT * FROM source WHERE date = today;
      → Rerun = same result (idempotent)

BETTER: MERGE/UPSERT with unique key
        → Handles partial failures too

2. Exactly-Once Semantics (Or At-Least-Once + Dedup)

True exactly-once is hard. Most pipelines use at-least-once delivery with deduplication:

Process record with id "abc"
  → Check: already processed?
    YES → skip
    NO  → process, mark "abc" as processed

3. Checkpointing

Record progress so failures don't restart from the beginning.

Process 10,000 records in batches of 1,000
  Batch 1: records 1-1000 → processed → checkpoint: 1000
  Batch 2: records 1001-2000 → processed → checkpoint: 2000
  Batch 3: CRASH
  Restart: read checkpoint (2000), resume from record 2001

4. Schema Validation

Validate data shape before processing. Catch problems at the source, not after loading garbage.

For each record:
  → Required fields present?
  → Types correct? (dates are dates, numbers are numbers)
  → Values in range? (age > 0, email has @)
  → Referential integrity? (user_id exists)

Invalid records → dead letter queue / error table (don't drop them silently)

Pipeline Design

Structure

┌──────────┐   ┌──────────────┐   ┌──────────┐   ┌──────────┐
│  Extract  │ → │  Validate    │ → │ Transform│ → │   Load   │
│ (source)  │   │ (schema/data)│   │ (business)│   │ (target) │
└──────────┘   └──────────────┘   └──────────┘   └──────────┘
       ↓              ↓                                ↓
   [audit log]   [error queue]                    [audit log]

Extract

• Read from source with explicit boundaries (date range, offset, cursor)
• Handle pagination (don't assume all data fits in memory)
• Record what was extracted (for audit and replay)
• Handle source unavailability gracefully (retry, alert, don't crash)

Validate

• Schema validation (types, required fields)
• Data quality checks (ranges, formats, referential integrity)
• Route invalid records to error queue (don't silently drop)
• Log validation metrics (% valid, common error types)

Transform

• Pure functions where possible (input → output, no side effects)
• Handle NULL/missing data explicitly
• Normalize formats (dates, currencies, encodings)
• Dedup if source may contain duplicates

Load

• Upsert / merge for idempotency
• Batch writes for performance
• Transactional where possible (all-or-nothing per batch)
• Verify record counts (expected vs loaded)

Failure Handling

| Failure | Strategy | |---------|----------| | Source unavailable | Retry with backoff, alert after N failures, run next scheduled attempt | | Invalid record | Route to error queue, continue processing valid records | | Transform error | Log with full context, route to error queue, continue | | Target unavailable | Retry with backoff, buffer locally if possible | | Partial batch failure | Checkpoint last successful batch, retry from there | | Duplicate delivery | Idempotent writes (upsert by unique key) |

Dead Letter Queue

Records that can't be processed go to a DLQ, not the void.

DLQ record:
  - Original record data
  - Error message
  - Pipeline name and step that failed
  - Timestamp
  - Retry count

Monitor DLQ size. Alert if > 0. Build tooling to inspect, fix, and replay DLQ records.

Backfill Strategy

When you need to reprocess historical data (new pipeline, schema change, bug fix):

1. DON'T rerun the full pipeline against all historical data at once
2. DO process in date-range batches (one day/week at a time)
3. Verify each batch before moving to the next
4. Run in parallel with the live pipeline (don't stop live processing)
5. Mark backfilled records (so you can distinguish them in analysis)

Backfill requirements:

• Pipeline MUST be idempotent (same input → same output regardless of run count)
• Pipeline MUST accept date range parameters (not just "today")
• Target MUST support upsert (not just insert)

Monitoring

| Metric | Alert Condition | |--------|----------------| | Records processed | Significantly fewer than expected (source went silent?) | | Error rate | > X% of records failing validation/transform | | Pipeline duration | > 2x normal (performance regression?) | | DLQ depth | > 0 (something is consistently failing) | | Data freshness | Target data older than SLA (pipeline not running?) | | Source-target count mismatch | Extracted ≠ loaded (data loss?) |

Data Quality Checks

Run after loading, before declaring success:

-- Record count matches
SELECT count(*) FROM target WHERE batch_id = 'X';
-- vs expected from extract step

-- No unexpected NULLs in required fields
SELECT count(*) FROM target WHERE required_field IS NULL AND batch_id = 'X';

-- Values in expected range
SELECT count(*) FROM target WHERE amount  1;

-- Freshness
SELECT max(updated_at) FROM target;
-- Should be recent

Anti-Patterns

| Anti-Pattern | Problem | Fix | |-------------|---------|-----| | No idempotency | Reruns create duplicates | Upsert by unique key, or delete-and-reinsert | | Silent data loss | Invalid records dropped, no one notices | Error queue + monitoring | | No checkpointing | Failure = restart from beginning | Record progress, resume from last checkpoint | | Memory-bound | OOM on large datasets | Stream/batch processing, don't load all into memory | | No monitoring | Pipeline silently stops, stale data | Alert on freshness, record count, error rate | | Hardcoded schedule | Can't backfill, can't run ad-hoc | Parameterize date range, support manual triggers |

Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

• Author: pvnarp
• Source: pvnarp/agent-skills
• License: MIT

Install and usage instructions live in the source repository linked above.