etl_pipelines v1.0.0

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.

1. 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.

1. 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.

1. 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.

1. 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.

1. 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.