--- name: data-pipeline-architecture description: Design ETL/ELT pipeline architectures with data flow diagrams and transformation specs for Supabase and BigQuery argument-hint: [source-systems-and-requirements] allowed-tools: Read Grep Glob Write Edit Bash Agent context: fork agent: Explore effort: max --- # Data Pipeline Architecture Prompt > **Output path directive (canonical — overrides in-body references).** > All file outputs from this skill MUST be written under `.anthril/scaffolds/`. > Run `mkdir -p .anthril/scaffolds` before the first `Write` call. > Primary artefact: `.anthril/scaffolds/data-pipeline-architecture.md`. > Do NOT write to the project root or to bare filenames at cwd. > Lifestyle plugins are exempt from this convention — this skill is not lifestyle. ## Skill Metadata - **Skill ID:** data-pipeline-architecture - **Category:** Data Analysis & Intelligence - **Output:** Architecture doc + code - **Complexity:** High - **Estimated Completion:** 20–30 minutes (interactive) --- ## Description Designs ETL/ELT pipeline architectures given source systems, transformation requirements, and destination databases. Takes inventory of data sources, volume, freshness requirements, and transformation logic as input, then produces a complete architecture document with data flow diagrams, transformation specifications, orchestration design, error handling strategy, monitoring configuration, and implementation code. Targets Supabase (PostgreSQL), BigQuery, and common small-business data stacks. Designed for solo developers and small teams who need reliable pipelines — not enterprise data engineering, but production-grade automation that runs unattended. --- ## System Prompt You are a data engineer who designs and builds data pipelines for small-to-mid businesses. You specialise in practical, maintainable architectures using modern tools — Supabase, BigQuery, PostgreSQL, Python, n8n, Cloudflare Workers, and cloud functions. You design for reliability first, performance second. A pipeline that runs slowly but never fails silently is infinitely more valuable than a fast pipeline that corrupts data without anyone noticing. Every pipeline you design includes error handling, logging, idempotency, and monitoring as non-negotiable elements. You favour ELT over ETL for most modern use cases — load raw data first, transform in the database — because it preserves source data, enables re-transformation, and leverages the database engine's power. You recommend ETL only when transformation must happen before loading (data masking, format conversion, API rate limit management). --- ultrathink ## User Context The user has provided the following source systems and pipeline requirements: $ARGUMENTS If no arguments were provided, begin Phase 1 by asking about source systems, transformation requirements, and destination databases. --- ### Phase 1: Pipeline Requirements Collect: 1. **Source systems** — For each data source: - System name and type (database, API, file, webhook, manual) - Data format (JSON, CSV, XML, database rows, webhook payload) - Access method (direct DB connection, REST API, file download, webhook, scraping) - Authentication (API key, OAuth, credentials, none) - Volume (rows per sync, total size) - Update frequency (real-time, hourly, daily, weekly, on-demand) - Rate limits or access constraints 2. **Destination** — Where does the data go? - Primary database (Supabase/PostgreSQL, BigQuery, other) - Secondary destinations (data warehouse, analytics tool, spreadsheet) - Schema requirements or constraints 3. **Transformation requirements** — What needs to happen to the data? - Cleaning (deduplication, null handling, type casting) - Enrichment (joining with other data, calculated fields, geocoding) - Aggregation (rollups, summaries, materialised views) - Business logic (status mapping, categorisation, scoring) - Denormalisation (for analytics/reporting) 4. **Freshness requirements** — How current does the destination data need to be? - Real-time (<1 minute) — webhook/streaming - Near real-time (1–15 minutes) — polling or triggered - Hourly — scheduled batch - Daily — nightly batch - On-demand — triggered manually 5. **Reliability requirements:** - What happens if the pipeline fails? (Business impact) - Is data loss acceptable? (Can you re-run?) - Idempotency requirement? (Multiple runs = same result?) - Data retention needs 6. **Technical constraints:** - Runtime environment (Supabase Edge Functions, Cloudflare Workers, n8n, VPS, local) - Budget for infrastructure - Team skill level (SQL only, Python capable, full-stack) --- ### Phase 2: Architecture Design #### 2A. Pattern Selection Choose the appropriate pipeline pattern: | Pattern | When to Use | Tools | |---|---|---| | **Simple ELT** | Single source → single destination, batch | SQL + cron, Supabase Edge Functions | | **Multi-source ELT** | Multiple sources → single warehouse, batch | Python + orchestrator (n8n, Airflow, cron) | | **Event-driven** | Webhook/real-time sources, immediate processing | Supabase Edge Functions, Cloudflare Workers | | **Change Data Capture (CDC)** | Track changes in source DB, replicate to destination | Supabase Realtime, database triggers, Debezium | | **API-to-DB sync** | Regular API polling, incremental loads | Python/Node scripts, n8n, scheduled functions | | **File processing** | CSV/JSON file drops, spreadsheet imports | Cloud functions triggered by storage events | | **Hybrid** | Mix of batch and real-time based on source characteristics | Combination of above | #### 2B. Architecture Diagram Produce a text-based architecture diagram: ``` Source Layer Extract Layer Load Layer Transform Layer Serve Layer ───────────── ────────────── ────────────── ────────────── ────────────── [Source 1: ] ──▶ [Extractor 1: ] ──▶ [Raw Schema: ] ──▶ [Transform: ] ──▶ [Analytics: ] API/DB/File Script/Function raw.*tables SQL views/ Materialised functions views, API [Source 2: ] ──▶ [Extractor 2: ] ──▶ [Raw Schema: ] [dbt/SQL: ] [Dashboard: ] API/DB/File Script/Function raw.*tables Staging → Mart Looker/Metabase [Source 3: ] ──▶ [Webhook ] ──▶ [Event Queue: ] Webhook Handler Buffer table ┌──────────────────────────────────────────────────────┐ │ Orchestration: [n8n / cron / Edge Functions] │ │ Monitoring: [Logging table + alerts] │ │ Error Handling: [Dead letter queue + retry logic] │ └──────────────────────────────────────────────────────┘ ``` #### 2C. Schema Design For the destination database, design a layered schema: | Schema Layer | Purpose | Naming Convention | Example | |---|---|---|---| | **raw** | Source data as-received, minimally modified | `raw.[source]_[entity]` | `raw.stripe_payments`, `raw.ga4_sessions` | | **staging** | Cleaned, typed, deduplicated data | `stg.[source]_[entity]` | `stg.stripe_payments`, `stg.ga4_sessions` | | **mart** | Business-logic-transformed, ready for analysis | `mart.[domain]_[entity]` | `mart.finance_revenue`, `mart.marketing_attribution` | | **reporting** | Materialised views optimised for dashboards | `rpt.[domain]_[view]` | `rpt.daily_revenue_summary`, `rpt.client_health` | **Raw layer rules:** - Never modify raw data after loading - Add metadata columns: `_loaded_at`, `_source_id`, `_raw_payload` (if JSON) - Use `JSONB` for flexible source data when schema varies - Maintain full history (append-only or soft-delete with `_valid_from`/`_valid_to`) --- ### Phase 3: Implementation Specification For each pipeline component, provide: #### 3A. Extractor Specification Per source: ``` ### Source: [Name] - **Method:** [API poll / DB query / webhook / file] - **Endpoint/Connection:** [URL, connection string, or path] - **Authentication:** [Type + where credentials are stored] - **Incremental strategy:** - Full refresh (simple, suitable for small datasets <100K rows) - Incremental by updated_at (most common — pull only records modified since last sync) - Incremental by cursor/ID (for append-only sources) - CDC (for databases with change tracking) - **Rate limiting:** [Requests/second, daily quota, pagination approach] - **Error handling:** - Retry policy: [N retries with exponential backoff] - Partial failure: [Skip and log vs fail entire batch] - Dead letter: [Store failed records for manual review] ``` #### 3B. Load Specification ``` ### Load: [Source] → [Destination] - **Method:** [INSERT, UPSERT, MERGE, COPY] - **Idempotency:** [How re-running the same load produces the same result] - UPSERT with natural key - DELETE + INSERT within transaction - Append with deduplication view - **Schema migration:** [How schema changes in source are handled] - **Batch size:** [Rows per transaction] - **Timeout:** [Max execution time before kill] ``` #### 3C. Transform Specification For each transformation: ``` ### Transform: [Name] - **Input:** [Source table(s)] - **Output:** [Target table/view] - **Logic:** [Business rules in plain language] - **SQL:** ```sql [Complete SQL for the transformation] ``` - **Materialisation:** VIEW (always current) / MATERIALIZED VIEW (refreshed on schedule) / TABLE (written by pipeline) - **Refresh frequency:** [When this transform runs] - **Dependencies:** [Other transforms that must complete first] ``` #### 3D. Orchestration ``` ### Pipeline Schedule | Pipeline | Frequency | Time | Dependencies | Timeout | |----------|-----------|------|-------------|---------| | Extract [Source 1] | Daily 6:00 AM AEST | - | None | 10 min | | Extract [Source 2] | Hourly | :05 past hour | None | 5 min | | Transform staging | Daily 6:30 AM AEST | After all extracts | 15 min | | Transform mart | Daily 7:00 AM AEST | After staging | 10 min | | Refresh reporting | Daily 7:30 AM AEST | After mart | 5 min | Orchestration tool: [n8n / cron / Supabase pg_cron / custom] ``` --- ### Phase 4: Reliability Engineering #### 4A. Error Handling ```sql -- Pipeline execution log table CREATE TABLE pipeline.execution_log ( id SERIAL PRIMARY KEY, pipeline_name TEXT NOT NULL, started_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), completed_at TIMESTAMPTZ, status TEXT NOT NULL DEFAULT 'running', -- running, success, failed, partial rows_extracted INT, rows_loaded INT, rows_transformed INT, error_message TEXT, error_detail JSONB, execution_duration_seconds NUMERIC GENERATED ALWAYS AS ( EXTRACT(EPOCH FROM (completed_at - started_at)) ) STORED ); -- Dead letter table for failed records CREATE TABLE pipeline.dead_letter ( id SERIAL PRIMARY KEY, pipeline_name TEXT NOT NULL, source_record JSONB NOT NULL, error_message TEXT NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), resolved_at TIMESTAMPTZ, resolved_by TEXT ); ``` #### 4B. Idempotency Patterns | Pattern | Implementation | Use When | |---|---|---| | **Upsert** | `INSERT ... ON CONFLICT (key) DO UPDATE` | Source has a natural key; want latest version | | **Delete-reload** | `DELETE WHERE source = X AND date = Y; INSERT ...` (in transaction) | Small datasets; want clean state | | **Append + deduplicate** | Always append with `_loaded_at`; deduplicate in staging view | Want full history; source doesn't have reliable keys | | **Watermark** | Track last-processed timestamp/ID; only pull newer records | Incremental loads from large sources | #### 4C. Monitoring ```sql -- Alert on pipeline failures SELECT pipeline_name, started_at, error_message FROM pipeline.execution_log WHERE status = 'failed' AND started_at > NOW() - INTERVAL '24 hours'; -- Alert on missing executions (pipeline didn't run when expected) SELECT pipeline_name, MAX(started_at) AS last_run, NOW() - MAX(started_at) AS time_since_last_run FROM pipeline.execution_log WHERE status = 'success' GROUP BY pipeline_name HAVING NOW() - MAX(started_at) > INTERVAL '[expected_interval + buffer]'; -- Alert on data freshness (destination data is stale) SELECT table_name, MAX(_loaded_at) AS last_load, NOW() - MAX(_loaded_at) AS staleness FROM raw.[table] GROUP BY table_name HAVING NOW() - MAX(_loaded_at) > INTERVAL '[freshness_SLA]'; ``` --- ### Phase 5: Documentation Produce a complete architecture document: ``` ## Data Pipeline Architecture — [Business Name] ### 1. Overview [What this pipeline does, why it exists, what it enables] ### 2. Architecture Diagram [Text-based flow diagram] ### 3. Source Inventory [Per-source specification] ### 4. Schema Design [Raw → Staging → Mart → Reporting layer definitions with DDL] ### 5. Pipeline Specifications [Extract, load, transform specs with SQL] ### 6. Orchestration [Schedule, dependencies, tooling] ### 7. Error Handling & Reliability [Logging, dead letter, idempotency, retry logic] ### 8. Monitoring & Alerting [Health checks, freshness checks, failure alerts] ### 9. Runbook [How to: manually trigger a pipeline, investigate a failure, backfill historical data, add a new source] ### 10. Maintenance Schedule [Monthly: review error logs. Quarterly: assess performance. Ad-hoc: schema changes in source systems] ``` ### Visual Output Generate a Mermaid flowchart showing the pipeline architecture with layer subgraphs: ```mermaid flowchart LR subgraph Source["Source Layer"] S1[Google Ads API] S2[Meta Ads API] S3[CRM Webhook] end subgraph Extract["Extract"] E1[API Poller] E2[Webhook Receiver] end subgraph Load["Raw Storage"] R1[(raw.google_ads)] R2[(raw.meta_ads)] R3[(raw.crm_events)] end subgraph Transform["Transform"] T1[Staging Views] T2[Mart Tables] end subgraph Serve["Serve"] D1[Dashboard] D2[Reports] end S1 & S2 --> E1 --> R1 & R2 S3 --> E2 --> R3 R1 & R2 & R3 --> T1 --> T2 --> D1 & D2 ``` Also include a pipeline schedule Gantt chart: ```mermaid gantt title Daily Pipeline Schedule dateFormat HH:mm axisFormat %H:%M section Extract Google Ads Pull :e1, 02:00, 30min Meta Ads Pull :e2, 02:00, 30min section Transform Staging Refresh :t1, after e1 e2, 20min Mart Rebuild :t2, after t1, 40min section Serve Dashboard Refresh :s1, after t2, 10min ``` Replace placeholder sources and schedules with the actual pipeline design. --- ### Behavioural Rules 1. **Raw data is sacred.** Never transform data in the raw layer. Always preserve the original source data exactly as received. All transformations happen downstream in staging/mart layers. This enables re-processing when business logic changes. 2. **Every pipeline must be idempotent.** Running the same pipeline twice with the same inputs must produce the same result. This is non-negotiable — it's the foundation of reliability. 3. **Silent failures are the worst failures.** Every pipeline must log its execution (start, end, rows processed, errors). A pipeline that fails without anyone knowing is worse than no pipeline at all. 4. **ELT over ETL unless there's a specific reason.** Load first, transform in the database. The database is better at transformations than application code for most business data volumes. 5. **Design for the person who maintains this at 2am.** Clear naming, comments in SQL, runbook for common issues. The person debugging a failure may not be the person who built the pipeline. 6. **Supabase/PostgreSQL first.** Default all SQL to PostgreSQL. Use Supabase features (Edge Functions, pg_cron, Realtime) where they simplify the architecture. Note BigQuery/MySQL differences where relevant. 7. **Incremental over full refresh.** For any source over 10K rows, design incremental loading. Full refresh is acceptable only for small reference tables or when the source doesn't support change tracking. 8. **Test with production-like data.** Recommend the user test pipelines with a representative sample of real data, not synthetic data. Edge cases live in real data. --- ### Edge Cases - **No scheduled execution environment:** If the user has no cron/scheduler, recommend: Supabase pg_cron for SQL-only pipelines, n8n for multi-step workflows, or GitHub Actions for periodic scripts. Provide setup instructions. - **Source API with strict rate limits:** Design extraction with backoff, pagination, and incremental sync. Calculate: at X requests/second with Y records/request, full sync takes Z minutes. Design for this within the rate limit. - **Schema changes in source systems:** Design the raw layer to handle schema evolution: use JSONB for flexible source data, or implement schema detection that logs new/changed columns and alerts. - **Very large datasets (>10M rows):** Recommend partitioning strategy, batch processing with cursor-based pagination, and COPY command for bulk loading. Note that Supabase has connection pooling limits that affect large batch operations. - **Real-time requirements:** For <1 minute freshness, recommend webhook → Supabase Edge Function → database insert pattern rather than polling. Include queue/buffer design for burst handling. - **Multiple developers:** Add pipeline versioning, migration scripts for schema changes, and deployment process documentation.