data-eng-data-quality

Data Engineering: Data Quality

Comprehensive data quality validation, monitoring, and observability patterns for production data pipelines.

Core Principles

1. Validate early, validate often -- Catch data issues at ingestion before they propagate downstream. Every pipeline stage should have quality gates.
2. Schema contracts are APIs -- Treat your data schemas as versioned contracts between producers and consumers. Breaking changes require coordination.
3. Measure the six dimensions -- Track completeness, accuracy, consistency, timeliness, uniqueness, and validity as quantifiable metrics with thresholds.
4. Observability over monitoring -- Move beyond threshold alerts to understanding data behavior through freshness, volume, schema, and lineage tracking.
5. Quality is a pipeline, not a step -- Data quality is not a single validation checkpoint; it is a continuous process woven into every stage of your data lifecycle.

Great Expectations Fundamentals

Define expectations as declarative rules, organize them into suites, and run checkpoints in your pipeline.

import great_expectations as gx

context = gx.get_context()
ds = context.data_sources.add_pandas("customer_source")
asset = ds.add_dataframe_asset(name="customers")
batch_def = asset.add_batch_definition_whole_dataframe("full_batch")

suite = context.suites.add(gx.ExpectationSuite(name="customer_ingestion_suite"))
suite.add_expectation(gx.expectations.ExpectColumnValuesToNotBeNull(column="customer_id"))
suite.add_expectation(gx.expectations.ExpectColumnValuesToBeUnique(column="customer_id"))
suite.add_expectation(gx.expectations.ExpectColumnValuesToMatchRegex(
    column="email", regex=r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$"
))
suite.add_expectation(gx.expectations.ExpectColumnValuesToBeBetween(
    column="account_balance", min_value=0, max_value=10_000_000
))

val_def = context.validation_definitions.add(
    gx.ValidationDefinition(name="customer_validation", data=batch_def, suite=suite)
)
checkpoint = context.checkpoints.add(
    gx.Checkpoint(name="customer_checkpoint", validation_definitions=[val_def])
)
result = checkpoint.run()
if not result.success:
    raise ValueError(f"Quality failed: {sum(1 for r in result.run_results.values() if not r.success)} checks")