Data Pipeline Architecture

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

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