Automated Data Quality with Great Expectations

Bad data is worse than no data. Downstream consumers—analysts, ML models, dashboards—make decisions based on your data. If it’s wrong, those decisions are wrong. In this chapter, we’ll implement automated data quality validation using Great Expectations 1.x, catching issues at ingestion rather than during analysis.

Why Data Quality Matters

Real-world AIS data contains all sorts of problems:

Issue	Example	Impact
Invalid MMSI	999999999 (test beacon)	Corrupts vessel tracking
Impossible coordinates	Latitude 91.0	Breaks geospatial queries
Teleporting vessels	Ship jumps 1000km in 1 second	Route analysis fails
Invalid speeds	500 knots	Anomaly detection triggers
Missing timestamps	NULL	Can’t order events

Without validation, analysts spend 80% of their time cleaning data. ML models learn from noise. Dashboards show wrong metrics. Debugging takes hours because nobody knows where bad data came from.

With validation: issues are caught early, rules are documented, alerts fire when quality degrades, and downstream consumers can trust the data.

Great Expectations 1.x Overview

Great Expectations (GX) is a Python framework for data validation. Version 1.x introduces a streamlined, code-first API—no CLI required.

Core Concepts

Component	Purpose
Data Context	Entry point for all GX operations
Expectation Suite	Collection of validation rules
Validation Definition	Links data to an expectation suite
Checkpoint	Runs validations and triggers actions
Data Docs	Auto-generated HTML reports

The flow: define expectations → connect to data → run validation → review results.

Setting Up Great Expectations

Install the library:

pip install great-expectations==1.10.0

Create a Data Context:

import great_expectations as gx

# Ephemeral context (in-memory)
context = gx.get_context()

# Persistent context (saves to disk)
context = gx.get_context(
    mode="file",
    project_root_dir="./great_expectations"
)

Creating Expectation Suites

An expectation suite is a collection of validation rules. Here’s a comprehensive suite for AIS silver layer data:

import great_expectations as gx

def create_silver_ais_suite(context: gx.DataContext) -> gx.ExpectationSuite:
    """Create expectation suite for silver AIS data."""
    suite = gx.ExpectationSuite(name="silver_ais_quality")

    # Schema: required columns must be present
    suite.add_expectation(
        gx.expectations.ExpectTableColumnsToMatchSet(
            column_set=[
                "timestamp", "mmsi", "latitude", "longitude",
                "speed_over_ground", "course_over_ground",
                "heading", "source", "dt"
            ],
            exact_match=False,  # Allow additional columns
        )
    )

    # Null checks: critical fields must not be null
    for column in ["timestamp", "mmsi", "latitude", "longitude"]:
        suite.add_expectation(
            gx.expectations.ExpectColumnValuesToNotBeNull(column=column)
        )

    # Range validation: MMSI (ITU standard: 200M-799M)
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeBetween(
            column="mmsi",
            min_value=200000000,
            max_value=799999999,
        )
    )

    # Range validation: coordinates
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeBetween(
            column="latitude",
            min_value=-90.0,
            max_value=90.0,
        )
    )

    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeBetween(
            column="longitude",
            min_value=-180.0,
            max_value=180.0,
        )
    )

    # Range validation: speed (0-102.2 knots, allow 5% outliers)
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeBetween(
            column="speed_over_ground",
            min_value=0.0,
            max_value=102.2,
            mostly=0.95,
        )
    )

    # Categorical: source must be known
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeInSet(
            column="source",
            value_set=["dma"],
        )
    )

    # Volume: reasonable row count per partition
    suite.add_expectation(
        gx.expectations.ExpectTableRowCountToBeBetween(
            min_value=10000,
            max_value=50000000,
        )
    )

    # Uniqueness: should have many distinct vessels
    suite.add_expectation(
        gx.expectations.ExpectColumnUniqueValueCountToBeBetween(
            column="mmsi",
            min_value=1000,
            max_value=100000,
        )
    )

    return context.suites.add(suite)

The mostly Parameter

Real data is never perfect. The mostly parameter lets you set acceptable thresholds:

gx.expectations.ExpectColumnValuesToBeBetween(
    column="speed",
    min_value=0,
    max_value=100,
    mostly=0.95  # 95% of values must pass
)

This prevents false positives from occasional sensor errors while still catching systematic issues.

Running Validation

Create a Validation Definition

A validation definition links your data to an expectation suite:

def validate_dataframe(df, context, suite_name="silver_ais_quality"):
    """Validate a DataFrame against the expectation suite."""

    # Get the suite
    suite = context.suites.get(suite_name)

    # Create data source for in-memory data
    data_source = context.data_sources.add_pandas(name="validation_source")
    data_asset = data_source.add_dataframe_asset(name="data")
    batch_definition = data_asset.add_batch_definition_whole_dataframe(
        name="batch"
    )

    # Create validation definition
    validation_def = gx.ValidationDefinition(
        name="validate_data",
        data=batch_definition,
        suite=suite,
    )

    # Run validation
    result = validation_def.run(batch_parameters={"dataframe": df})

    return result

Interpreting Results

result = validate_dataframe(df, context)

# Overall success
print(f"Validation passed: {result.success}")

# Statistics
stats = result.results[0].result.get("statistics", {})
evaluated = stats.get("evaluated_expectations", 0)
successful = stats.get("successful_expectations", 0)
success_rate = (successful / evaluated * 100) if evaluated > 0 else 0

print(f"Success rate: {success_rate:.1f}%")

# Failed expectations
for exp_result in result.results:
    if not exp_result.success:
        print(f"Failed: {exp_result.expectation_config.type}")

Dagster Integration

Integrate validation into your pipeline as a Dagster asset:

from dagster import asset, AssetIn, Output, MetadataValue

@asset(
    partitions_def=daily_partition,
    ins={"dma_ais_silver": AssetIn()},
    description="Validate DMA silver layer data quality"
)
def dma_silver_validation(context, dma_ais_silver, minio):
    """Validate silver layer with Great Expectations.

    Blocks downstream gold layer if validation fails.
    """
    partition_date = context.partition_key

    # Read silver data
    df = read_silver_data(minio, f"silver/ais/source=dma/dt={partition_date}/")
    context.log.info(f"Loaded {len(df)} rows for validation")

    # Initialize GX
    gx_context = gx.get_context()

    # Get or create suite
    try:
        suite = gx_context.suites.get("silver_ais_quality")
    except Exception:
        suite = create_silver_ais_suite(gx_context)

    # Run validation
    data_source = gx_context.data_sources.add_pandas(name=f"dma_{partition_date}")
    data_asset = data_source.add_dataframe_asset(name="silver_data")
    batch_def = data_asset.add_batch_definition_whole_dataframe(name="batch")

    validation_def = gx.ValidationDefinition(
        name=f"validate_dma_{partition_date}",
        data=batch_def,
        suite=suite,
    )

    result = validation_def.run(batch_parameters={"dataframe": df})

    # Extract metrics
    stats = result.results[0].result.get("statistics", {})
    evaluated = stats.get("evaluated_expectations", 0)
    successful = stats.get("successful_expectations", 0)
    success_percent = (successful / evaluated * 100) if evaluated > 0 else 0

    # Fail if validation fails
    if not result.success:
        raise ValueError(f"Validation failed ({success_percent:.1f}% success)")

    return Output(
        value=True,
        metadata={
            "success_percent": MetadataValue.float(success_percent),
            "evaluated_expectations": MetadataValue.int(evaluated),
            "failed_expectations": MetadataValue.int(evaluated - successful),
            "row_count": MetadataValue.int(len(df)),
        }
    )

Pipeline with Validation Gate

Your asset dependency graph now includes validation as a gate:

dma_ais_bronze → dma_ais_silver → dma_silver_validation → gold assets

If validation fails, gold layer processing is blocked. Bad data never reaches downstream consumers.

Handling Validation Failures

Strategy 1: Fail Fast (Recommended)

Block downstream processing if validation fails:

if not result.success:
    raise ValueError("Validation failed - blocking downstream")

Pros: Ensures only quality data reaches consumers Cons: Pipeline stops when data degrades

Strategy 2: Conditional Blocking

Fail on critical expectations, warn on others:

critical_types = [
    "expect_column_values_to_not_be_null",
    "expect_column_values_to_be_between",
]

for exp_result in result.results:
    if not exp_result.success:
        if exp_result.expectation_config.type in critical_types:
            raise ValueError(f"Critical failure: {exp_result}")
        else:
            context.log.warning(f"Non-critical failure: {exp_result}")

Strategy 3: Warn and Continue

Log warnings but continue processing:

if not result.success:
    context.log.warning(f"Quality degraded: {success_percent:.1f}%")
# Continue anyway

Recommendation: Start strict, relax selectively based on experience.

Data Docs

Great Expectations auto-generates HTML documentation:

# Build data docs
context.build_data_docs()

# Open in browser
context.open_data_docs()

Data Docs show:

• All expectations with pass/fail status
• Validation history over time
• Detailed failure information
• Data profiling statistics

Best Practices

Start Simple

Don’t create 100 expectations on day 1:

1. Start with schema + null checks
2. Add range checks after observing data
3. Add custom expectations for known issues

Document Expectations

Explain why each expectation exists:

suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeBetween(
        column="mmsi",
        min_value=200000000,
        max_value=799999999,
        meta={
            "notes": "Valid MMSI range per ITU-R M.1371 standard",
            "impact": "Invalid MMSIs break vessel tracking"
        }
    )
)

Monitor Quality Over Time

Track validation metrics in your monitoring system:

metrics = {
    "partition_date": partition_date,
    "success_percent": success_percent,
    "failed_expectations": failed_count,
}
# Export to Prometheus/Grafana

Alert on Degradation

# Prometheus alert rule
- alert: DataQualityDegraded
  expr: data_quality_success_percent < 95
  for: 1h
  annotations:
    summary: "Data quality below 95%"

Common Expectations Reference

# Schema
gx.expectations.ExpectTableColumnsToMatchSet(
    column_set=["a", "b", "c"]
)

# Nulls
gx.expectations.ExpectColumnValuesToNotBeNull(column="mmsi")

# Ranges
gx.expectations.ExpectColumnValuesToBeBetween(
    column="lat", min_value=-90, max_value=90
)

# Sets
gx.expectations.ExpectColumnValuesToBeInSet(
    column="source", value_set=["dma"]
)

# Uniqueness
gx.expectations.ExpectColumnValuesToBeUnique(column="id")

# Row count
gx.expectations.ExpectTableRowCountToBeBetween(
    min_value=1000, max_value=1000000
)

# Unique value count
gx.expectations.ExpectColumnUniqueValueCountToBeBetween(
    column="mmsi", min_value=100, max_value=50000
)

What You’ve Built

Your pipeline now includes automated data quality validation that:

• Validates schema, nulls, ranges, and volumes
• Blocks bad data from reaching gold layer
• Documents quality rules as code
• Generates human-readable reports
• Integrates seamlessly with Dagster

Data quality checks should be automated, comprehensive, and integrated into your pipeline. Not a manual afterthought.

What’s Next

With validated silver layer data, Chapter 8 covers creating curated gold layer datasets—aggregating silver data into business-ready outputs like vessel tracks and daily statistics.

---

Building data quality into your pipeline? We help organizations implement production-grade validation. Get in touch to discuss your project.