When this skill is activated, always start your first response with the 🧢 emoji.

Data Quality

Data quality is the practice of ensuring that data is accurate, complete, consistent, timely, and trustworthy as it flows through pipelines and systems. Without explicit quality gates, bad data propagates silently - corrupting dashboards, training flawed models, and breaking downstream consumers. This skill covers the five pillars: schema validation at ingress, expectation-based testing with Great Expectations, data contracts between producers and consumers, lineage tracking for impact analysis, and continuous monitoring for anomaly detection.

When to use this skill

Trigger this skill when the user:

• Adds data validation or schema enforcement to a pipeline (ingestion, transformation, or serving)
• Writes Great Expectations expectation suites or checkpoints
• Defines data contracts between a producer team and consumer teams
• Implements data lineage tracking or impact analysis
• Sets up data quality monitoring dashboards or freshness/volume alerts
• Investigates data quality incidents (missing columns, null spikes, schema drift)
• Profiles a new dataset to understand distributions and anomalies
• Builds row-count, freshness, or distribution-based quality checks

Do NOT trigger this skill for:

• General ETL/ELT pipeline orchestration (use an Airflow/dbt skill instead)
• Data modeling or warehouse design decisions without a quality focus

Key principles

1. Validate at boundaries, not in the middle - Enforce quality at ingestion (before data enters your warehouse) and at serving (before consumers read it). Validating mid-pipeline catches problems too late to prevent downstream damage and too early to catch transformation bugs.

2. Contracts are APIs for data - A data contract is a formal agreement between a producer and consumer on schema, semantics, SLAs, and ownership. Treat it like a versioned API - breaking changes require migration paths, not surprise emails.

3. Test data like you test code - Every table should have expectations that run on every pipeline execution. Column nullability, uniqueness constraints, value ranges, referential integrity, and freshness are not optional - they are the unit tests of data engineering.

4. Lineage enables impact analysis - You cannot assess the blast radius of a schema change without knowing what reads from what. Instrument lineage at the query level (not just table level) so you can trace column-level dependencies.

5. Monitor trends, not just thresholds - A row count of 1M is fine today but means nothing without historical context. Use statistical anomaly detection (z-score, moving averages) to catch gradual drift that static thresholds miss.

Core concepts

The five dimensions of data quality

Data contracts

A data contract defines: the schema (column names, types, constraints), semantic meaning (what "revenue" means - gross or net), SLAs (freshness, volume bounds), and ownership (who to page when it breaks). Contracts are versioned artifacts stored alongside code - not wiki pages that rot. The producer owns the contract and is responsible for not shipping breaking changes without a version bump.

Data lineage

Lineage is a directed acyclic graph (DAG) where nodes are datasets (tables, views, files) and edges are transformations (SQL queries, Spark jobs, dbt models). Column-level lineage tracks which output columns derive from which input columns. Tools like OpenLineage, DataHub, and dbt's built-in lineage provide this automatically when integrated into your orchestrator.

Great Expectations

Great Expectations (GX) is the standard open-source framework for data testing. The core abstractions are: Data Source (connection to your data), Expectation Suite (a collection of assertions about a dataset), Validator (runs expectations against data), and Checkpoint (an orchestratable unit that validates data and triggers actions on pass/fail). Expectations are declarative - expect_column_values_to_not_be_null - and produce rich, human-readable validation results.

Common tasks

Write a Great Expectations suite

Define expectations for a table covering nullability, types, ranges, and uniqueness.

import great_expectations as gx

context = gx.get_context()

# Connect to data source
datasource = context.data_sources.add_postgres(
    name="warehouse",
    connection_string="postgresql+psycopg2://user:pass@host:5432/db",
)
data_asset = datasource.add_table_asset(name="orders", table_name="orders")
batch_definition = data_asset.add_batch_definition_whole_table("full_table")

# Create expectation suite
suite = context.suites.add(
    gx.ExpectationSuite(name="orders_quality")
)

suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="order_id")
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeUnique(column="order_id")
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeBetween(
        column="total_amount", min_value=0, max_value=1_000_000
    )
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeInSet(
        column="status", value_set=["pending", "completed", "cancelled", "refunded"]
    )
)
suite.add_expectation(
    gx.expectations.ExpectTableRowCountToBeBetween(min_value=1000, max_value=10_000_000)
)

Always start with not-null and uniqueness expectations on primary keys before adding business-logic expectations.

Run a checkpoint in a pipeline

Wire a Great Expectations checkpoint into your orchestrator so validation runs on every load.

import great_expectations as gx

context = gx.get_context()

# Define a checkpoint that validates the orders suite
checkpoint = context.checkpoints.add(
    gx.Checkpoint(
        name="orders_checkpoint",
        validation_definitions=[
            gx.ValidationDefinition(
                name="orders_validation",
                data=context.data_sources.get("warehouse")
                    .get_asset("orders")
                    .get_batch_definition("full_table"),
                suite=context.suites.get("orders_quality"),
            )
        ],
        actions=[
            gx.checkpoint_actions.UpdateDataDocsAction(name="update_docs"),
        ],
    )
)

# Run in Airflow task / dbt post-hook / standalone script
result = checkpoint.run()
if not result.success:
    failing = [r for r in result.run_results.values() if not r.success]
    raise RuntimeError(f"Data quality check failed: {len(failing)} validations failed")

Define a data contract

Create a YAML contract between a producer and consumer team.

# contracts/orders-v2.yaml
apiVersion: datacontract/v1.0
kind: DataContract
metadata:
  name: orders
  version: 2.0.0
  owner: payments-team
  consumers:
    - analytics-team
    - ml-team

schema:
  type: table
  database: warehouse
  table: public.orders
  columns:
    - name: order_id
      type: string
      constraints: [not_null, unique]
      description: UUID primary key
    - name: customer_id
      type: string
      constraints: [not_null]
      description: FK to customers.customer_id
    - name: total_amount
      type: decimal(10,2)
      constraints: [not_null, gte_0]
      description: Gross order