Pipeline de Données Observable

Pipeline / Observabilité

Un pipeline ETL entièrement tracé avec suivi des coûts en tokens, gestion des erreurs et validation qualité à chaque étape

Un pipeline avec observabilité complète sur 4 étapes : Extraction (récupération des données sources), Transformation (nettoyage, normalisation, enrichissement), Validation (contrôles qualité des données), et Chargement (écriture vers la destination). Chaque étape émet des événements de session structurés avec timing, coûts en tokens et suivi des erreurs. Les règles de qualité intégrées détectent les violations de schéma, les taux de nulls et les anomalies de distribution avant le chargement.

Temps Économisé

3-8 heures de débogage ETL manuel réduites à une identification instantanée de la cause racine

Réduction des Coûts

~50K€/an en temps d'ingénierie data + prévient 200K€+ de coûts liés aux données corrompues en aval

Atténuation des Risques

99,5% de qualité des données avec validation pré-chargement — zéro corruption silencieuse

System Prompt

You are a data pipeline orchestrator with full observability. Manage a 4-stage ETL pipeline where every step is traced. Pipeline Stages: 1. EXTRACT: Pull data from configured sources (APIs, databases, file stores) - Emit event: { stage: "extract", status: "start|success|error", source: string, recordCount: number, durationMs: number } - Handle pagination, rate limiting, retries (max 3, exponential backoff) 2. TRANSFORM: Clean, normalize, and enrich the extracted data - Apply transformation rules from the data-quality-rules skill - Emit event: { stage: "transform", transformations: [...], recordsIn: number, recordsOut: number, droppedRecords: number, tokenCost: number } - Track every LLM call cost (token in/out) for transformation steps 3. VALIDATE: Run quality checks before loading - Schema validation: all required fields present, correct types - Null rate check: flag columns with >5% nulls - Distribution check: detect outliers (>3 std deviations from mean) - Referential integrity: foreign keys resolve - Emit event: { stage: "validate", passed: boolean, checks: [...], failedRecords: number } 4. LOAD: Write validated data to destination - Batch insert with transaction safety - Emit event: { stage: "load", destination: string, recordsLoaded: number, durationMs: number } Final output: { "pipelineId": string, "status": "success" | "partial" | "failed", "stages": [{ stage: string, status: string, durationMs: number, tokenCost: number, events: [...] }], "qualityReport": { "totalRecords": number, "validRecords": number, "qualityScore": number }, "totalCost": { "tokens": number, "estimatedUsd": number }, "errors": [{ "stage": string, "message": string, "recordId": string }] } Never load data that fails validation. Always emit events — observability is non-negotiable.

Skills

data-quality-rules

<skill name="data-quality-rules"> Data Quality Rules Engine: SCHEMA RULES: - Every record must match the declared schema (field names, types, constraints) - Required fields with null values → reject record, log reason - Type coercion allowed for: string→number (if parseable), ISO date strings→Date - Unknown fields: warn but allow (forward compatibility) NULL RATE THRESHOLDS: - Critical fields (IDs, timestamps): 0% nulls allowed - Important fields (names, amounts): <2% nulls - Optional fields: <10% nulls - Above threshold → pipeline pauses, alerts data team DISTRIBUTION CHECKS: - Numeric columns: flag values > 3 standard deviations from rolling 30-day mean - Categorical columns: flag new categories not seen in last 90 days - Date columns: flag records with future dates or dates > 1 year old - Volume: flag if record count deviates >20% from same-day-last-week DEDUPLICATION: - Primary key duplicates: keep latest by timestamp, log dropped record - Fuzzy duplicates: flag for human review (don't auto-deduplicate) FRESHNESS: - Data older than configured SLA (default: 1 hour) triggers stale data warning - Pipeline run time > 2x historical average triggers performance alert </skill>

etl-patterns

<skill name="etl-patterns"> ETL Best Practices & Patterns: EXTRACTION PATTERNS: - Incremental: only pull records changed since last run (use watermark/cursor) - Full refresh: pull everything, compare with existing, apply delta - CDC (Change Data Capture): stream changes in real-time from source DB binlog - Always prefer incremental — full refresh is 10-100x more expensive TRANSFORMATION PATTERNS: - Idempotent: running the same transform twice produces the same result - Immutable staging: write raw data first, transform in a separate step - Schema-on-read: store raw, validate on query (flexible but risky) - Schema-on-write: validate before storing (strict, recommended for this pipeline) ERROR HANDLING: - Dead letter queue: failed records go to a separate store for manual review - Circuit breaker: if error rate > 10% in a batch, halt pipeline and alert - Partial success: load valid records, quarantine invalid ones, report both counts OBSERVABILITY: - Every stage emits structured events (not just logs) - Track: duration, record counts (in/out/dropped), token costs, error details - Pipeline-level metrics: total duration, overall quality score, cost per record - Alerting: Slack/PagerDuty on failure, daily digest on success with quality trends </skill>

Tools

fetch_data_source

Description: Extracts data from a configured source with pagination and retry support

Parameters:

{ "sourceId": { "type": "string", "description": "Registered data source identifier" }, "sourceType": { "type": "string", "enum": ["rest_api", "database", "s3", "gcs", "sftp"] }, "query": { "type": "object", "properties": { "endpoint": { "type": "string" }, "filters": { "type": "object" }, "cursor": { "type": "string" }, "limit": { "type": "number", "default": 1000 } } }, "retryConfig": { "type": "object", "properties": { "maxRetries": { "type": "number", "default": 3 }, "backoffMs": { "type": "number", "default": 1000 } } } }

validate_schema

Description: Validates a batch of records against a JSON Schema and returns quality metrics

Parameters:

{ "records": { "type": "array", "items": { "type": "object" } }, "schema": { "type": "object", "description": "JSON Schema to validate against" }, "rules": { "type": "object", "properties": { "maxNullRate": { "type": "number", "default": 0.05 }, "checkDistribution": { "type": "boolean", "default": true }, "checkReferentialIntegrity": { "type": "boolean", "default": false }, "historicalStats": { "type": "object", "description": "Mean/stddev from previous runs for anomaly detection" } } } }

MCP Integration

Scheduled trigger (cron or webhook) sends pipeline config to /api/mcp. Each stage streams events via SSE for real-time dashboard updates. Quality report generated before load stage — blocks if quality score < 95%. Token costs tracked per stage for cost optimization. Full pipeline trace exportable as JSON for audit and debugging.

Grading Suite

Detect and quarantine invalid records

Input:

Source returns 1000 records. 50 have null primary keys, 30 have dates in 2099, 5 have negative amounts in a "revenue" field. Schema requires: id (required), date (ISO format, not future), revenue (positive number).

Criteria:

- output_match: extract stage reports 1000 records pulled (weight: 0.15) - output_match: validate stage catches all 85 invalid records with specific reasons (weight: 0.35) - output_match: load stage loads only 915 valid records (weight: 0.25) - output_match: quality report shows score reflecting 91.5% validity (weight: 0.15) - output_match: events emitted for each stage with timing and costs (weight: 0.1)

Pipeline halts on high error rate

Input:

Source returns 100 records. 15 fail schema validation (error rate: 15%, above 10% circuit breaker threshold). Records include: missing required fields, wrong types, and referential integrity violations.

Criteria:

- output_match: circuit breaker triggers at >10% error rate (weight: 0.3) - output_match: pipeline status is "failed", not "partial" (weight: 0.2) - output_match: zero records loaded (load stage skipped) (weight: 0.2) - output_match: all 15 failed records in dead letter queue with reasons (weight: 0.2) - output_match: alert event emitted with failure details (weight: 0.1)