dltHub data quality features include metrics for monitoring dataset properties over time, and checks to validate them against expectations. Together, they offer visibility and allow to catch data issues early. Metrics and checks are defined via Python code. The extensive configuration allows you to specify what to monitor and validate, when, how, and where to store results.

This page covers the basics of metrics and checks. You should notice a lot of symmetry (for example, with_metrics() and with_checks()). The later parts of this page cover notions applicable to both.

Metrics#

A data quality metric or metric a function applied to data that returns a scalar value describing a property of the data. A metric can take as input a column, a table, or the full dataset (i.e., all tables and historical metrics).

Define metrics#

Static#

You can define metrics along your @dlt.resource (and @dlt.transformer, @dlt.hub.transformation) via the new decorator @with_metrics. It is available under the dlt.hub.data_quality module, commonly imported as dq. Inside the decorator, you can set the individual metrics available through dq.metrics.column., dq.metrics.table., or dq.metrics.dataset..

The next snippet defines 3 metrics on the customers resource: the mean of the amount column, the number of null values in the email column, and the total number of rows in the table.

import dlt
from dlt.hub import data_quality as dq

@dq.with_metrics(
    dq.metrics.column.mean("amount"),
    dq.metrics.column.null_count("email"),
    dq.metrics.table.row_count()
)
@dlt.resource
def customers():
    yield data

The next snippet shows how to add dataset-level metrics to a source. The total_row_count is added on the crm source.

import dlt
from dlt.hub import data_quality as dq

@dq.with_metrics(
    dq.metrics.column.mean("amount"),
    dq.metrics.column.null_count("email"),
    dq.metrics.table.row_count()
)
@dlt.resource
def customers():
    yield data

@dq.with_metrics(
    dq.metrics.dataset.total_row_count()
)
@dlt.source
def crm():
    return [customers]

Dynamic#

Similar to the static approach, you can add metrics to an instantiated resource or source object using with_metrics. This is particularly useful when using built-in sources and resources like filesystem, rest_api or sql_database.

import dlt
from dlt.hub import data_quality as dq

@dlt.resource
def customers():
    yield from [...]

# later; this mutates the resource object and sets metrics
dq.with_metrics(
    customers,
    dq.metrics.column.mean("amount"),
    dq.metrics.column.null_count("email"),
    dq.metrics.table.row_count()
)

Available metrics#

Here's the list of built-in metrics:

from dlt.hub import data_quality as dq

# column-level
dq.metrics.column.maximum("col")
dq.metrics.column.minimum("col")
dq.metrics.column.mean("col")
dq.metrics.column.median("col")
dq.metrics.column.mode("col")
dq.metrics.column.sum("col")
dq.metrics.column.standard_deviation("col")
dq.metrics.column.quantile("col", quantile=0.95)
dq.metrics.column.null_count("col")
dq.metrics.column.null_rate("col")
dq.metrics.column.unique_count("col")
dq.metrics.column.average_length("col")
dq.metrics.column.minimum_length("col")
dq.metrics.column.maximum_length("col")

# table-level
dq.metrics.table.row_count() # Number of rows in table
dq.metrics.table.unique_count() # Number of distinct / unique rows in table
dq.metrics.table.null_row_count() # Number of rows where all columns are null

# dataset-level
dq.metrics.dataset.total_row_count() # Total number of rows
dq.metrics.dataset.load_row_count() # Rows added in latest load
dq.metrics.dataset.latest_loaded_at() # Timestamp of most recent load

Compute metrics#

Pass the pipeline to dq.enable_data_quality() to enable metrics. This will set a flag on the pipeline state to compute metrics and write results to destination after each pipeline.run() call.

pipeline = dlt.pipeline("my_pipeline", destination="duckdb")

dq.enable_data_quality(pipeline)

pipeline.run(customers)

Since the flag is stored on the pipeline state, instantiating this pipeline from another script or notebook will remember that metrics are enabled and compute metrics when pipeline.run() is called.

# another_file.py
pipeline = dlt.attach("my_pipeline")

# this will compute metrics too
pipeline.run(customers)

Read metrics#

The convenience function dq.read_metric() allows you to retrieve stored metrics with some metadata. This makes it easy to build reporting, dashboard, or analytics over this data.

The function produces a dlt.Relation which can be converted to a list, pandas dataframe, arrow table, etc.

dataset = pipeline.dataset()
# column-level `mean` as pandas.DataFrame
dq.read_metric(
    dataset, 
    table="customers", 
    column="amount", 
    metric="mean"
).df()

# table-level `row_count` as list of tuples
dq.read_metric(
    dataset, 
    table="customers", 
    metric="row_count"
).fetchall()

# dataset-level `total_row_count` as pyarrow.Table
dq.read_metric(
    dataset, 
    metric="total_row_count"
).arrow()

Checks#

A data quality check or check is a function applied to data that returns a check result or result (can be boolean, integer, float, etc.). The result that is converted to a success / fail check outcome or outcome (boolean) based on a decision.

Define checks#

Static#

You can define checks along your @dlt.resource (and @dlt.transformer, @dlt.hub.transformation) via the new decorator @with_checks available under the dlt.hub.data_quality module. Inside the decorator, you can set the individual checks available through dq.checks..

This snippet shows a single is_in() check being ran against the orders table.

import dlt
from dlt.hub import data_quality as dq

@dq.with_checks( # type: ignore[attr-defined]
    dq.checks.is_in("payment_methods", ["card", "cash", "voucher"]),
)
@dlt.resource
def orders():
    yield from [...]

Dynamic#

Similar to the static approach, you can add checks to an instantiated resource or source object using with_checks. This is particularly useful when using built-in sources and resources like filesystem, rest_api or sql_database.

import dlt
from dlt.hub import data_quality as dq

@dlt.resource
def orders():
    yield data

# later; this mutates the resource object and sets checks
dq.with_checks(
    orders,
    dq.checks.is_in("payment_methods", ["card", "cash", "voucher"]),
)

Available checks#

Here's the list of built-in checks:

from dlt.hub import data_quality as dq

dq.checks.is_unique("col")
dq.checks.is_not_null("col")
dq.checks.is_primary_key("col") # valid primary key
dq.checks.is_in("foo", ["bar", "baz"]) # valid values
dq.checks.case("col < 0") # row-wise check

Compute checks#

Pass the pipeline to dq.enable_data_quality() to enable checks. This will set a flag on the pipeline state to compute checks and write results to destination after each pipeline.run() call.

pipeline = dlt.pipeline("my_pipeline", destination="duckdb")

dq.enable_data_quality(pipeline)

pipeline.run(customers)

Since the flag is stored on the pipeline state, instantiating this pipeline from another script or notebook will remember that metrics are enabled and compute checks when pipeline.run() is called.

# another_file.py
pipeline = dlt.attach("my_pipeline")

# this will compute checks too
pipeline.run(customers)

Read checks#

The convenience function dq.read_check() allows you to retrieve stored checks with some metadata. This makes it easy to build reporting, dashboard, or analytics over this data.

The function produces a dlt.Relation which can be converted to a list, pandas dataframe, arrow table, etc.

dataset = pipeline.dataset()
dq.read_check(
    dataset,
    table="orders",
    column="payment_method", 
).df()

Lifecycle#

Data quality (both metrics and checks) can be executed at different stages of the pipeline lifecycle. This impacts several aspects including:

• available input data
• compute resources used
• actions available after a failed check (for example, prevent invalid data load)

Post-load#

The post-load execution is the simplest option. The pipeline goes through Extract -> Normalize -> Load as usual. Then, the checks are executed on the destination.

Properties:

• Failed records can't be dropped or quarantined before load. All records must be written, checked, and then handled. This only works with write_disposition="append" or destinations supporting snapshots (for example iceberg, ducklake).
• Checks have access to the full dataset. This includes current and past loads + internal dlt tables.
• Computed directly on the destination. This scales well with the size of the data and the complexity of the checks.
• Results and outcome are directly stored on the dataset. No data movement is required.

Pre-load (staging)#

The pre-load execution via staging dataset allows you to execute checks on the destination and trigger actions before data is loaded into the dataset. This is effectively using post-load checks before a second load phase.

Properties:

• Failed records can be dropped or quarantined before load. This works with all write_disposition
• Requires a destination that supports staging datasets.
• Checks have access to the current load.
  • If the staging dataset is on the same destination, checks can access the full dataset.
  • If the staging dataset is on a different destination, communication between the staging dataset and the dataset.
• Computed on the staging destination. This scales well with the size of the data and the complexity of the checks.
• Data and checks results & outcome can be safely stored on the staging dataset until review. This helps human-in-the-loop workflows without reprocessing the full pipeline.

Pre-load (in-memory)#

The pre-load execution in-memory will execute checks using duckdb against the load packages (i.e., temporary files) stored on the machine that runs dlt. This allows you to trigger actions before data is loaded into the destination.

Properties:

• Failed records can be dropped or quarantined before load. This works with all `write_disposition
• Checks only have access to the current load. Checking against the full dataset requires communication between the staging destination and the main destination.
• Computed on the machine running the pipeline. The resource need to match the compute requirements.
• Data and checks results & outcome may be lost if the runtime is ephemeral (for example, AWS Lambda timeout). In this case, the pipeline must process the data again.

Roadmap#

• Define checks that depend on metrics (this should reduce verbosity)
• Support user-defined group-by metrics
• Support completely custom checks via @dlt.hub.transformation (for example, SQL, SQLGlot, Ibis, Narwhals, Polars)
• Trigger actions based on check result or outcome (for example, send Slack notification)
• Track metrics and checks changes via dlt.Schema versioning