---

name: data-pipeline-architecture description: Implements data pipeline architectures (batch ETL/ELT, streaming, medallion bronze-silver-gold layers, data quality gates, schema evolution handling) for reliable data processing at scale. license: MIT compatibility: opencode archetypes:

• tactical
• orchestration anti_triggers:
• brainstorming
• vague ideation
• long-form architecture response_profile: verbosity: low directive_strength: high abstraction_level: operational metadata: version: "1.0.0" domain: coding triggers: data pipeline, ETL architecture, batch processing, stream processing, data quality gates, schema evolution, Spark pipeline, how do i build a data pipeline role: implementation scope: implementation output-format: code content-types: [code, guidance, config, do-dont, examples] related-skills: event-driven-architecture, system-reliability-architecture, distributed-systems-architecture, database-design-patterns

---

Data Pipeline Architecture

Implements data pipeline architectures that reliably ingest, transform, and serve data across batch and streaming workloads. This skill makes the model design fault-tolerant pipelines with proper data quality enforcement, schema evolution handling, and observable failure recovery.

TL;DR Checklist

• Choose batch, streaming, or hybrid based on freshness requirements (not technology preference)
• Implement medallion architecture: bronze (raw) → silver (cleaned/validated) → gold (business aggregates)
• Add data quality gates at every stage with explicit pass/fail handling
• Handle schema evolution explicitly — never assume the source schema stays static
• Design idempotent transformations for exactly-once semantics on reprocessing
• Implement dead letter queue for records that fail validation
• Add lineage tracking to record which transforms produced each output record

---

When to Use

• Designing a new data platform or analytics pipeline from scratch
• Migrating a legacy ETL system (cron scripts, SQL-only) to a modern batch/streaming pipeline
• Implementing medallion architecture (bronze/silver/gold layers) for structured data governance
• Adding data quality enforcement to an existing pipeline that processes untrusted sources
• Handling schema changes from upstream systems without breaking downstream consumers
• Building real-time or near-real-time streaming pipelines with Kafka, Flink, or Spark Structured Streaming

---

When NOT to Use

• Simple one-off data extraction scripts — use a single Spark job or SQL query instead
• OLTP database design — this covers analytical/data platform architecture, not transactional systems
• Real-time application serving (user-facing APIs) — use api-architecture or cloud-native-architecture instead
• Machine learning model training pipeline orchestration — focus only on data movement/transform, not model lifecycle

---

Core Workflow

1. Define Data Contract — Identify source systems, ingestion frequency, schema expectations, and SLA requirements for freshness. Checkpoint: Document the schema as a typed contract (field names, types, nullability, constraints) that both producer and consumer agree on.

2. Select Processing Mode — Choose batch, streaming, or hybrid based on data volume velocity, and freshness requirements:

   • Volume > 1TB/day or T+1 reporting → batch-first (Spark, dbt)
   • Latency < 5 minutes required → streaming (Kafka Streams, Flink)
   • Mixed workloads with different SLAs → hybrid (batch for historical, streaming for real-time) Checkpoint: Do not select streaming because it is "modern" — batch is simpler and cheaper for most workloads.

3. Design Medallion Layers — Implement the three-tier architecture:

   • Bronze: Raw append-only storage, exact source fidelity, partitioned by ingestion time
   • Silver: Cleaned, deduplicated, typed data with quality gates applied; conformed to common schemas
   • Gold: Business-level aggregates optimized for analytics, denormalized where needed Checkpoint: Each layer must have its own schema registry and quality rules — never skip a layer.

4. Implement Quality Gates — Add validation at bronze→silver and silver→gold transitions:

   • Not-null constraints on key fields
   • Type enforcement with explicit casting or rejection
   • Range checks (dates in the future, negative quantities)
   • Referential integrity against reference datasets Checkpoint: Define explicit policies for failed records — quarantine to dead letter queue vs. auto-fix with defaults.

5. Handle Schema Evolution — Implement forward/backward compatibility:

   • Always add fields as nullable (forward compatible)
   • Never remove fields that downstream consumers may read (backward compatible)
   • Use schema registry with explicit versioning
   • Maintain a field deprecation timeline before dropping columns Checkpoint: Test every schema migration against existing consumer pipelines before promoting to production.

6. Add Observability — Implement monitoring at every stage:

   • Row count comparison between layers (bronze count ≥ silver count after quality filtering)
   • Late data handling with watermark configuration for streaming
   • Alert on quality gate failure rate exceeding threshold
   • End-to-end latency tracking from source to gold

---

Implementation Patterns

Pattern 1: Medallion Architecture with Spark Structured Streaming

"""Medallion architecture layers — bronze (raw) → silver (cleaned) → gold (aggregated).

Each layer has explicit quality gates and idempotent processing.
"""
from pyspark.sql import SparkSession
from pyspark.sql.functions import (
    col, current_timestamp, to_timestamp, when, lit, countDistinct
)
from pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType


def read_bronze(spark: SparkSession, table_name: str) -> "pyspark.DataFrame":
    """Read raw data from bronze layer — append-only, exact source fidelity."""
    return spark.read.table(f"bronze.{table_name}")


def apply_quality_gates(
    df: "pyspark.DataFrame",
    required_fields: list[str],
    quality_threshold: float = 0.95,
) -> tuple["pyspark.DataFrame", int]:
    """Validate records and return clean dataframe + count of rejected records.

    Args:
        df: Input dataframe (typically from bronze layer).
        required_fields: Field names that must not be null for a record to pass.
        quality_threshold: Minimum fraction of records that must pass quality gates.

    Returns:
        Tuple of (cleaned_df, rejected_count). Raises if rejection rate exceeds threshold.
    """
    # Mark each record as passing or failing quality checks
    df_with_flag = df.withColumn(
        "_quality_status",
        when(
            all(col(f).isNotNull() for f in required_fields),
            lit("pass"),
        ).otherwise(lit("fail"))
    )

    total_count = df_with_flag.count()
    rejected_count = df_with_flag.filter(col("_quality_status") == "fail").count()
    pass_rate = (total_count - rejected_count) / max(total_count, 1)

    if pass_rate < quality_threshold:
        raise ValueError(
            f"Quality gate failed: {pass_rate:.2%} pass rate "
            f"(rejected {rejected_count}/{total_count}). "
            f"Investigate source data before reprocessing."
        )

    cleaned = df_with_flag.filter(col("_quality_status") == "pass").drop("_quality_status")
    return cleaned, rejected_count


def write_silver(
    spark: SparkSession,
    table_name: str,
    df: "pyspark.DataFrame",
) -> None:
    """Write cleaned data to silver layer with idempotent overwrite by partition."""
    df.write.mode("overwrite").partitionBy("ingestion_date").saveAsTable(f"silver.{table_name}")


def build_gold_aggregates(
    silver_df: "pyspark.DataFrame",
    group_keys: list[str],
    metrics: dict[str, str],
) -> "pyspark.DataFrame":
    """Build gold layer aggregates from silver data.

    Args:
        silver_df: Cleaned silver-layer dataframe.
        group_keys: Columns to group by for aggregation.
        metrics: Mapping of output column name → aggregate expression string.
            Example: {"total_revenue": "SUM(revenue)", "avg_order_value": "AVG(order_value)"}

    Returns:
        Aggregated gold-layer dataframe.
    """
    from pyspark.sql import functions as F

    agg_exprs = [F.expr(expr).alias(col_name) for col_name, expr in metrics.items()]
    return silver_df.groupBy(*group_keys).agg(*agg_exprs)


def process_pipeline(
    spark: SparkSession,
    source_table: str,
    required_fields: list[str],
    group_keys: list[str],
    metrics: dict[str, str],
) -> None:
    """End-to-end medallion pipeline: bronze → silver → gold.

    Demonstrates the full data quality and transformation flow.
    """
    # Bronze to Silver: read raw, apply quality gates, write cleaned
    bronze_df = read_bronze(spark, source_table)
    silver_df, rejected_count = apply_quality_gates(
        bronze_df, required_fields=required_fields, quality_threshold=0.95
    )

    if rejected_count > 0:
        # In production, write rejected records to a dead letter queue table
        print(f"Rejected {rejected_count} records — check dead_letter.{source_table}")

    write_silver(spark, source_table, silver_df)

    # Silver to Gold: aggregate for analytics
    gold_df = build_gold_aggregates(silver_df, group_keys=group_keys, metrics=metrics)
    gold_df.write.mode("overwrite").saveAsTable(f"gold.{source_table}_daily")

Pattern 2: Schema Evolution Handler (BAD vs. GOOD)

"""Schema evolution patterns — forward and backward compatibility."""

# ❌ BAD: Fragile schema — breaking changes cascade downstream
def process_raw_data(raw_row: dict) -> dict:
    """Assumes exact source schema; breaks on any field change."""
    return {
        "user_id": raw_row["userId"],          # Renamed → KeyError
        "email": raw_row["email_address"],     # Removed → KeyError
        "timestamp": raw_row["created_at"],    # Type changed (str → datetime) → TypeError
        "amount": float(raw_row["amount"]),    # No null safety
    }

# ✅ GOOD: Tolerant schema parser with explicit versioning and safe defaults
from typing import Any, Optional
from dataclasses import dataclass, field


@dataclass
class SchemaVersion:
    """Tracks schema version and maps source fields to canonical names."""
    version: int
    field_mappings: dict[str, str]       # canonical_name → source_field_name
    required_fields: list[str]
    type_coercions: dict[str, type]      # canonical_name → expected_type
    default_values: dict[str, Any] = field(default_factory=dict)

    def resolve(self, raw_row: dict) -> Optional[dict]:
        """Safely map raw row to canonical schema with error handling."""
        result = {}
        for canonical_name, source_field in self.field_mappings.items():
            if source_field not in raw_row:
                # Field was removed — use default or skip
                if canonical_name in self.required_fields:
                    return None  # Required field missing → reject
                result[canonical_name] = self.default_values.get(
                    canonical_name, None
                )
            else:
                value = raw_row[source_field]
                try:
                    if value is not None and self.type_coercions.get(canonical_name):
                        result[canonical_name] = self.type_coercions[
                            canonical_name
                        ](value)
                    else:
                        result[canonical_name] = value
                except (TypeError, ValueError):
                    # Type coercion failed — reject this record
                    return None
        return result


# Example usage with schema v1 → v2 migration
SCHEMA_V1 = SchemaVersion(
    version=1,
    field_mappings={
        "user_id": "userId",
        "email": "email_address",
        "created_at": "created_at",
        "amount": "amount",
    },
    required_fields=["user_id", "amount"],
    type_coercions={"user_id": int, "amount": float},
)

# V2 adds optional fields; V1 schema still works (backward compatible)
SCHEMA_V2 = SchemaVersion(
    version=2,
    field_mappings={
        "user_id": "userId",              # Same as V1
        "email": "email_address",         # Same as V1
        "created_at": "created_at",       # Same as V1
        "amount": "amount",               # Same as V1
        "country_code": "country_code",   # New optional field
        "plan_tier": "plan_tier",         # New optional field
    },
    required_fields=["user_id", "amount"],  # No change — no breaking impact on V1 consumers
    type_coercions={"user_id": int, "amount": float},
    default_values={"country_code": "US", "plan_tier": "free"},
)


def process_with_schema_evolution(raw_row: dict, version: int) -> Optional[dict]:
    """Process a record using the appropriate schema version.

    Args:
        raw_row: Raw data from upstream source.
        version: Schema version embedded in the source metadata.

    Returns:
        Canonical data record, or None if validation fails.
    """
    schemas = {1: SCHEMA_V1, 2: SCHEMA_V2}
    schema = schemas.get(version)
    if schema is None:
        # Unknown schema version — send to dead letter queue for investigation
        raise ValueError(f"Unsupported schema version: {version}")

    resolved = schema.resolve(raw_row)
    return resolved

Pattern 3: Streaming Pipeline with Kafka and Spark Structured Streaming

"""Streaming pipeline pattern — Kafka source → transformation → sink.

Uses watermarking for late data handling and checkpointing for fault tolerance.
"""
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, window, to_timestamp


def create_streaming_pipeline(
    spark: SparkSession,
    bootstrap_servers: str,
    topic: str,
    group_id: str,
    checkpoint_dir: str,
) -> None:
    """Build a streaming pipeline with proper watermarking and fault tolerance.

    Args:
        bootstrap_servers: Kafka broker connection string.
        topic: Source Kafka topic to consume from.
        group_id: Consumer group for exactly-once semantics.
        checkpoint_dir: HDFS/S3 path for Spark checkpoint state.
    """
    # Read stream from Kafka with schema inference disabled for safety
    streaming_df = (
        spark.readStream.format("kafka")
        .option("kafka.bootstrap.servers", bootstrap_servers)
        .option("subscribe", topic)
        .option("group.id", group_id)
        .option("startingOffsets", "latest")
        .option("maxOffsetsPerTrigger", "100000")  # Backpressure control
        .load()
    )

    # Parse JSON value and add watermark for late event handling (30-minute window)
    parsed = streaming_df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")

    from pyspark.sql.functions import from_json, col as spark_col
    from pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType

    value_schema = StructType([
        StructField("event_type", StringType(), True),
        StructField("timestamp", StringType(), True),
        StructField("amount", DoubleType(), True),
        StructField("user_id", StringType(), True),
    ])

    enriched = parsed.withColumn(
        "parsed",
        from_json(spark_col("value"), value_schema)
    ).select(
        spark_col("parsed.*"),
        spark_col("key").alias("partition_key"),
    ).withColumn(
        "event_time",
        to_timestamp(col("timestamp"))
    )

    # Apply watermark: drop events older than 30 minutes from the latest event seen
    watermarked = enriched.withWatermark("event_time", "30 minutes")

    # Aggregate in 5-minute windows with output mode that tracks state changes
    windowed_agg = watermarked.groupBy(
        window(col("event_time"), "5 minutes"),
        col("event_type"),
    ).agg(
        spark_col("COUNT(*)").alias("event_count"),
        spark_col("SUM(col('amount'))").alias("total_amount"),
    )

    # Write to sink (Delta Lake table with upsert mode)
    query = (
        windowed_agg.writeStream
        .format("delta")
        .outputMode("update")
        .option("checkpointLocation", checkpoint_dir)
        .trigger(processingTime="1 minute")
        .start("silver.streaming_aggregates")
    )

    # Block until streaming query terminates
    query.awaitTermination()

---

Constraints

MUST DO

• Implement idempotent transformations so reprocessing produces identical results
• Use append-only storage at the bronze layer — never overwrite raw source data
• Maintain a schema registry with versioned schemas and explicit compatibility rules
• Add quality gates before every silver→gold transition with configurable thresholds
• Configure watermarks in streaming pipelines to handle late-arriving data
• Implement dead letter queues for records that fail validation — never silently drop them
• Track lineage: record which input files/records produced each output partition

MUST NOT DO

• Store processed/transformed data as the source of truth — bronze layer is always the source of truth
• Apply business logic at the bronze layer — keep raw data untouched and append-only
• Use streaming for all workloads — batch is simpler, cheaper, and sufficient for most analytical queries
• Bypass quality gates to "fix things faster" — a single bad record can corrupt downstream aggregations
• Remove or modify records in the silver layer after they pass quality gates — always add corrective transformations as new records
• Store schemas inline in code without versioning — schema drift will cause silent data corruption

---

Output Template

When designing a data pipeline, produce:

1. Pipeline Architecture Diagram — ASCII diagram showing sources → bronze → silver → gold → consumers, with processing mode (batch/stream/hybrid) labeled on each stage
2. Data Contract Specification — Table of source fields, canonical names, types, nullability, and constraints for the first 5 key tables
3. Quality Gate Definition — List of validation rules per layer with rejection handling policy
4. Schema Evolution Plan — Migration steps from current schema to target schema with backward/forward compatibility verification
5. Implementation Code — Python/PySpark code for the primary transformation logic

---

Related Skills

Skill	Purpose
event-driven-architecture	Event streaming infrastructure (Kafka, RabbitMQ) that powers data pipeline messaging
system-reliability-architecture	Resilience patterns (circuit breakers, retry strategies) for pipeline fault tolerance
distributed-systems-architecture	Consistency models and partitioning strategies for distributed data storage
database-design-patterns	Delta Lake/Iceberg/Hudi table formats that serve as the silver/gold storage layer

---

Live References

Authoritative documentation for data pipeline architecture patterns and tools. The model follows markdown links at load time to resolve external references and inline content.

• Apache Spark Structured Streaming Programming Guide
• Delta Lake Documentation
• Lambda vs Kappa Architecture Patterns
• Medallion Architecture (DataBricks)
• Apache Flink Streaming Analytics Documentation
• Data Quality Best Practices — Great Expectations
• Schema Registry Pattern — Confluent