Testing

End-to-End Data Pipeline Validation: From Ingestion to Serving Layer

07 Sep 2026 — 5 min read

Unit tests verify individual transformations. Schema tests catch bad data. But neither tells you whether the full pipeline — from raw ingestion through transformation to the serving layer — is correct. End-to-end validation catches the bugs that only appear when all the pieces interact: a join condition that drops rows, an aggregation with the wrong granularity, a schema evolution that silently coerces types.

This post covers strategies for validating data pipelines across their full lifecycle.

Reconciliation Testing: Verify Row Counts and Aggregates

The simplest and most effective end-to-end test: compare aggregates between the source and destination.

# tests/reconciliation/test_orders_pipeline.py
import pytest
from pyspark.sql import SparkSession
from datetime import date

@pytest.fixture(scope="session")
def spark():
    return SparkSession.builder.master("local[2]").getOrCreate()

@pytest.fixture
def run_date():
    return date(2026, 5, 1)

def test_order_count_matches_source(spark, run_date):
    """Row count from source should match the serving layer."""
    
    source_count = (spark.read.format("delta")
        .load("s3://bucket/raw/orders/")
        .filter(f"date(created_at) = '{run_date}'")
        .count())
    
    serving_count = (spark.read.format("delta")
        .load("s3://bucket/warehouse/fct_orders/")
        .filter(f"order_date = '{run_date}'")
        .count())
    
    # Allow 0.1% tolerance for deduplication
    discrepancy_pct = abs(source_count - serving_count) / source_count
    assert discrepancy_pct < 0.001, \
        f"Row count discrepancy: source={source_count}, serving={serving_count}"

def test_revenue_matches_source(spark, run_date):
    """Total revenue in serving layer should match raw orders."""
    
    source_revenue = (spark.read.format("delta")
        .load("s3://bucket/raw/orders/")
        .filter(f"date(created_at) = '{run_date}' AND status = 'completed'")
        .agg({"amount": "sum"})
        .collect()[0][0])
    
    serving_revenue = (spark.read.format("delta")
        .load("s3://bucket/warehouse/fct_revenue_daily/")
        .filter(f"revenue_date = '{run_date}'")
        .agg({"net_revenue": "sum"})
        .collect()[0][0])
    
    # Revenue should match within $1 (rounding tolerance)
    assert abs(source_revenue - serving_revenue) < 1.0, \
        f"Revenue discrepancy: source=${source_revenue:.2f}, serving=${serving_revenue:.2f}"

Data Lineage Validation

Verify that data flows through the expected transformations:

# tests/lineage/test_customer_pipeline_lineage.py

def test_all_source_customers_appear_in_serving(spark):
    """Every customer in the source should appear in the customer dimension."""
    
    source_customer_ids = set(
        row.customer_id for row in 
        spark.read.format("delta")
            .load("s3://bucket/raw/customers/")
            .select("customer_id")
            .collect()
    )
    
    serving_customer_ids = set(
        row.customer_id for row in
        spark.read.format("delta")
            .load("s3://bucket/warehouse/dim_customers/")
            .filter("is_current = true")
            .select("customer_id")
            .collect()
    )
    
    missing = source_customer_ids - serving_customer_ids
    assert len(missing) == 0, \
        f"{len(missing)} customers from source missing in serving layer: {list(missing)[:10]}"

def test_no_phantom_customers_in_serving(spark):
    """Serving layer should not contain customer IDs that don't exist in source."""
    
    source_ids = set(row.customer_id for row in 
        spark.read.format("delta").load("s3://bucket/raw/customers/")
            .select("customer_id").collect())
    
    serving_ids = set(row.customer_id for row in
        spark.read.format("delta").load("s3://bucket/warehouse/dim_customers/")
            .select("customer_id").collect())
    
    phantom = serving_ids - source_ids
    assert len(phantom) == 0, \
        f"{len(phantom)} phantom customers in serving layer"

Schema Evolution Testing

Schema changes are the most common cause of silent pipeline failures. Test that your pipeline handles them gracefully:

# tests/schema/test_schema_evolution.py
from pyspark.sql.types import StructType, StructField, StringType, LongType

def test_pipeline_handles_new_optional_column(spark, tmp_path):
    """Adding a new nullable column to source should not break the pipeline."""
    
    # Write source data with new column
    new_schema_data = [
        (1, "customer_1", "completed", 100.0, "NEW_FIELD_VALUE"),
        (2, "customer_2", "pending", 50.0, None),
    ]
    source_df = spark.createDataFrame(
        new_schema_data,
        "order_id LONG, customer_id STRING, status STRING, amount DOUBLE, new_field STRING"
    )
    source_df.write.format("parquet").save(str(tmp_path / "source"))
    
    # Pipeline should process without errors
    from myetl.pipelines.orders import run_orders_pipeline
    
    try:
        result = run_orders_pipeline(str(tmp_path / "source"), str(tmp_path / "output"))
        assert result.count() > 0, "Pipeline produced no output"
    except Exception as e:
        pytest.fail(f"Pipeline failed on new column: {e}")

def test_pipeline_fails_on_removed_required_column(spark, tmp_path):
    """Removing a required column should produce a clear error, not silent wrong output."""
    
    # Write source data without required column
    broken_data = [(1, "customer_1", 100.0)]  # missing 'status' column
    source_df = spark.createDataFrame(
        broken_data, "order_id LONG, customer_id STRING, amount DOUBLE"
    )
    source_df.write.format("parquet").save(str(tmp_path / "source"))
    
    from myetl.pipelines.orders import run_orders_pipeline
    
    with pytest.raises(Exception) as exc_info:
        run_orders_pipeline(str(tmp_path / "source"), str(tmp_path / "output"))
    
    assert "status" in str(exc_info.value).lower(), \
        "Error message should mention the missing column"

def test_type_coercion_does_not_corrupt_data(spark, tmp_path):
    """String 'amount' should be properly parsed, not silently nulled."""
    
    # Source sends amount as string (e.g., from JSON)
    source_data = [("1", "100.50"), ("2", "invalid_amount"), ("3", "75.00")]
    source_df = spark.createDataFrame(source_data, "order_id STRING, amount STRING")
    source_df.write.format("parquet").save(str(tmp_path / "source"))
    
    from myetl.pipelines.orders import run_orders_pipeline
    result = run_orders_pipeline(str(tmp_path / "source"), str(tmp_path / "output"))
    
    # Valid amounts should be preserved
    valid_rows = result.filter("amount IS NOT NULL")
    assert valid_rows.count() == 2, "Valid amounts should be parsed correctly"
    
    # Invalid amount should be flagged, not silently nulled
    invalid_rows = result.filter("amount IS NULL")
    assert invalid_rows.count() == 1
    assert invalid_rows.first()["parse_error"] is not None, \
        "Invalid amounts should have a parse_error column set"

Data Contract Testing

Data contracts define the interface between producer and consumer. Test that producers don't break consumers:

# tests/contracts/test_orders_contract.py
import json
from pathlib import Path

# Load the contract (shared between producer and consumer teams)
CONTRACT = json.loads(Path("contracts/orders_v2.json").read_text())

def test_orders_output_satisfies_contract(spark):
    """The orders pipeline output must conform to the v2 contract."""
    
    df = spark.read.format("delta").load("s3://bucket/warehouse/fct_orders/")
    
    # Check required columns exist
    for col_def in CONTRACT["required_columns"]:
        assert col_def["name"] in df.columns, \
            f"Required column '{col_def['name']}' missing from output"
        
        actual_type = dict(df.dtypes)[col_def["name"]]
        expected_type = col_def["type"]
        assert actual_type == expected_type, \
            f"Column '{col_def['name']}' type changed: expected {expected_type}, got {actual_type}"
    
    # Check no nulls in non-nullable columns
    for col_def in CONTRACT["required_columns"]:
        if not col_def.get("nullable", True):
            null_count = df.filter(f"{col_def['name']} IS NULL").count()
            assert null_count == 0, \
                f"Non-nullable column '{col_def['name']}' has {null_count} nulls"
    
    # Check value constraints
    for constraint in CONTRACT.get("value_constraints", []):
        violations = df.filter(f"NOT ({constraint['expression']})").count()
        assert violations == 0, \
            f"Constraint '{constraint['name']}' violated: {violations} rows"

// contracts/orders_v2.json
{
  "name": "orders",
  "version": 2,
  "required_columns": [
    {"name": "order_id",    "type": "bigint",  "nullable": false},
    {"name": "customer_id", "type": "bigint",  "nullable": false},
    {"name": "amount",      "type": "double",  "nullable": true},
    {"name": "status",      "type": "string",  "nullable": false},
    {"name": "order_date",  "type": "date",    "nullable": false}
  ],
  "value_constraints": [
    {"name": "positive_amount",  "expression": "amount IS NULL OR amount >= 0"},
    {"name": "valid_status",     "expression": "status IN ('pending','confirmed','shipped','cancelled')"},
    {"name": "not_future_date",  "expression": "order_date <= current_date()"}
  ]
}

Time-Travel Testing for Incremental Pipelines

Delta Lake's time travel lets you test that incremental runs produce the same result as full refreshes:

def test_incremental_matches_full_refresh(spark):
    """Incremental updates should produce the same final state as a full refresh."""
    
    # Get state from full refresh (run once a week)
    full_refresh_state = (spark.read.format("delta")
        .option("versionAsOf", "latest")
        .load("s3://bucket/warehouse/fct_orders/")
        .orderBy("order_id"))
    
    # Get state from incremental (runs daily)
    incremental_state = (spark.read.format("delta")
        .load("s3://bucket/warehouse/fct_orders_incremental/")
        .orderBy("order_id"))
    
    assert full_refresh_state.count() == incremental_state.count(), \
        "Full refresh and incremental have different row counts"
    
    # Deep comparison (expensive — run only in nightly CI)
    full_revenue = full_refresh_state.agg({"amount": "sum"}).collect()[0][0]
    incr_revenue = incremental_state.agg({"amount": "sum"}).collect()[0][0]
    
    assert abs(full_revenue - incr_revenue) < 0.01, \
        f"Revenue discrepancy: full={full_revenue}, incremental={incr_revenue}"

CI Pipeline

# .github/workflows/pipeline-e2e-tests.yml
name: Pipeline E2E Tests
on:
  schedule:
    - cron: '0 6 * * *'  # daily, after overnight pipeline runs
  workflow_dispatch:

jobs:
  reconciliation:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      - run: pip install pyspark pytest
      - name: Run reconciliation tests
        run: pytest tests/reconciliation/ -v
        env:
          AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
          AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
  
  contracts:
    runs-on: ubuntu-latest
    needs: reconciliation  # only run if reconciliation passes
    steps:
      - uses: actions/checkout@v4
      - run: pip install pyspark pytest
      - name: Validate data contracts
        run: pytest tests/contracts/ -v
        env:
          AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
          AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}

End-to-end pipeline validation is not a substitute for unit tests — it's the final layer that catches integration bugs. Run unit tests on every commit, reconciliation tests nightly, and schema/contract tests whenever the pipeline code changes. Each layer has a different cost and catches different bugs; together they give you confidence that what reaches your BI tools and ML models is actually correct.