---
name: "etl_pipelines"
description: "Designing ETL pipelines: extraction with pagination and checkpointing, idempotent transforms, upsert loads, cron vs event scheduling, row-count observability, and dead-letter handling. Triggers on: build an ETL pipeline, extract transform load, data pipeline, idempotent upsert, pipeline retries causing duplicates, dead letter queue for data, scheduled data sync."
metadata: { "includeInPrompt": true }
---

## Purpose

Design reliable extract-transform-load pipelines where the reference stack is plain Python, a clock (cron), and a database (sqlite/postgres) — no orchestration framework required. This skill covers the *orchestration* angle: extraction loops, idempotency, retries, and observability. It is not file repair (csv-cleaning), column wrangling (pandas-patterns), or contract testing (data-validation).

## Workflow

1. **Extract with pagination and checkpointing.** Page API results by cursor (preferred — offsets shift under concurrent writes) or offset, and persist a high-water mark (last cursor / timestamp) after each page. File ingestion reads in chunks and records the last processed byte or row offset. A crashed run resumes from the checkpoint, never from the start.

2. **Stage raw, then transform.** Keep two layers: a *raw* layer storing the source payload untouched, and a *clean* layer holding typed, normalized rows. Transforms must be pure functions of the raw record — same input, same output, no wall-clock or random state — so re-running a transform never corrupts clean data.

3. **Load with upserts on a stable key.** The business key (source record id) becomes a UNIQUE constraint; every load is `INSERT ... ON CONFLICT(source_id) DO UPDATE` (verified working on sqlite 3.45.1; postgres uses the same syntax). Schema-on-write: the load step enforces types and rejects surprises — never insert "just in case" columns.

4. **Schedule by freshness need.** Cron for periodic batch syncs; event-driven (webhook/queue trigger) only when near-real-time matters. Either way the pipeline must be re-runnable: schedulers double-fire and events redeliver, so idempotency is what makes scheduling safe, not the scheduler.

5. **Observe row counts, not vibes.** Every run logs `rows_in`, `rows_loaded`, `dead_letters`, and start/finish timestamps (e.g. an `etl_runs` table). Zero rows loaded is a signal, not a success — alert on empty runs and on dead-letter spikes. Malformed records go to a dead-letter store with the raw payload and the error, never silently dropped or silently coerced.

6. **Handle the two big failure modes.** (a) *Duplicate loads from retries*: killed by the upsert on a stable key — a retry overwrites, never appends. (b) *Partial writes*: a mid-run crash leaves half a load; commit per atomic row (upsert) or in one transaction per run, then retry to convergence. Verified end-to-end: 51 records extracted → 50 loaded + 1 dead letter; re-run → still 50 distinct rows, 0 duplicates; run killed after 25 rows then retried → 50 distinct rows, 0 duplicates.

## Output Contract

After applying this skill, an agent should produce:

- The pipeline shape: extract loop (cursor/checkpoint location), transform function (raw→clean mapping), load target (table + unique key).
- The upsert statement for the chosen database, with the uniqueness constraint named.
- A per-run observability schema (`etl_runs`: rows_in, rows_loaded, dead_letters, timestamps) and the dead-letter record format.
- The retry-safety argument in one sentence: which key makes a second run converge instead of duplicate.

## Operating Rules

- **Unique key before data.** No load step ships without a UNIQUE constraint on the business key; an `INSERT` without one is a duplicate waiting for a retry.
- **Never silently drop.** Malformed records land in dead letters with the raw payload and the error message; counts are logged every run.
- **Raw is immutable.** The raw layer is never edited in place — every fix lives in the transform, so a transform bug is re-runnable against history.
- **Idempotent by construction.** The pipeline must be safe to run twice back-to-back; prove it by re-running and comparing distinct row counts (my reference run: 50 → 50, zero duplicates).
- **Retry converges, never appends.** After any mid-run failure, re-run from the checkpoint and let the upsert absorb the overlap (verified: killed at 25/50 rows, retry converged to exactly 50).
- **Count everything.** A run that extracts 0 rows or loads 0 rows gets flagged, not celebrated — silence is the failure signal pipelines hide behind.
- **Cron is the default for batch.** Event-driven scheduling only when the source offers reliable delivery and the transform stays idempotent; the scheduler choice never substitutes for upsert safety.