Summary Metrics

Available since v0.10.0

DQX provides comprehensive functionality to capture and store aggregate statistics about your data quality. This allows you to track data quality trends over time, monitor the health of your data pipelines, and gain insights into the overall quality of your datasets. For the metrics table schema and relationships to output, quarantine, and checks tables, see Table Schemas and Relationships.

Overview

Summary metrics in DQX capture built-in metrics (automatically calculated), per-check metrics (error and warning counts broken down by check name), and custom metrics (user-defined SQL expressions) during data quality checking. These metrics are collected using Spark's built-in Observation functionality and can be persisted to tables for historical analysis.

Built-in Metrics

DQX automatically captures the following built-in metrics for every data quality check execution:

Metric Name	Data Type	Description
input_row_count	int	Total number of input rows processed
error_row_count	int	Number of rows that failed error-level checks
warning_row_count	int	Number of rows that triggered warning-level checks
valid_row_count	int	Number of rows that passed all checks (have no errors and warnings)
check_metrics	string (JSON)	Per-check breakdown of error and warning counts, stored as a JSON array of structs: [{"check_name": "...", "error_count": N, "warning_count": N}, ...]. Automatically included when checks are applied.

Per-Check Metrics

Available since v0.14.0

When checks are applied, DQX automatically includes a check_metrics metric that provides a per-check breakdown of error and warning counts. This makes it possible to identify which specific checks are failing without querying the row-level _errors and _warnings columns.

check_metrics contains an entry for every applied check, not only the ones that failed. Checks that never triggered still appear with error_count and warning_count set to 0. This differs from the row-level _errors / _warnings columns, which list only the checks that failed for a given row. As a result, the breakdown is derived from the applied checks (their names), so it cannot be reconstructed from the data alone — a check with zero violations leaves no trace in _errors / _warnings.

The check_metrics value is a JSON-serialized array of structs, with one entry per applied check (here passenger_incorrect_count was applied but never triggered):

[
  {"check_name": "id_is_not_null", "error_count": 5, "warning_count": 0},
  {"check_name": "name_is_not_null_and_not_empty", "error_count": 0, "warning_count": 3},
  {"check_name": "passenger_incorrect_count", "error_count": 0, "warning_count": 0}
]

Each entry contains:

• check_name — the name of the check (either explicitly set via name in the rule definition, or auto-derived from the check function and arguments)
• error_count — number of rows where this check triggered an error (0 if no rows failed the check)
• warning_count — number of rows where this check triggered a warning (0 if no rows failed the check)

When persisted to the metrics table, check_metrics is stored as a single row with metric_name = 'check_metrics' and metric_value containing the JSON string. You can parse it with Spark's from_json or json_tuple functions for analysis:

SELECT
  run_id,
  check.check_name,
  check.error_count,
  check.warning_count
FROM main.analytics.dq_metrics
LATERAL VIEW explode(
  from_json(metric_value, 'array<struct<check_name:string,error_count:bigint,warning_count:bigint>>')
) AS check
WHERE metric_name = 'check_metrics'

Custom Metrics

Users can define custom metrics with Spark SQL expressions. These metrics will be collected in addition to DQX's built-in metrics.

Avoid Expensive Operations

Summary metrics are calculated on all records processed by DQX. Complex aggregations can degrade performance when processing large datasets. Be cautious with operations like DISTINCT on high-cardinality columns.

Example of custom data quality summary metrics:

sum(array_size(_errors)) as total_errors
avg(array_size(_errors)) as errors_avg
count(case when array_size(_errors) > 1) as count_multiple_errors

See the Configuring Custom Metrics section for instructions on setting them up.

Programmatic approach

Accessing Metrics when Applying Checks

Engine methods for applying checks (e.g. apply_checks, apply_checks_by_metadata, apply_checks_and_split, apply_checks_by_metadata_and_split) can optionally return a Spark Observation with one or more output DataFrames. Data quality metrics can be accessed from the Spark Observation after any action is performed on the output DataFrames.

Streaming metrics

Metrics are not directly accessible from the returned Spark Observation when data is processed with streaming. Use DQX's built-in methods to persist streaming metrics to an output table. See Writing Metrics to a Table with Streaming for more details.

Python

from databricks.labs.dqx.engine import DQEngine
from databricks.labs.dqx.metrics_observer import DQMetricsObserver
from databricks.sdk import WorkspaceClient

# Create observer
observer = DQMetricsObserver(name="dq_metrics")

# Create the engine with the optional observer
engine = DQEngine(WorkspaceClient(), observer=observer)

# Apply checks and get metrics
checked_df, observation = engine.apply_checks_by_metadata(df, checks)

# Apply checks, split and get metrics
#valid_df, quarantine_df, observation = engine.apply_checks_by_metadata_and_split(df, checks)

# Trigger an action to populate metrics (e.g., count, save to a table).
# Without triggering an action, metrics will not be populated, and accessing them will result in a stall.
row_count = checked_df.count()

# Access metrics
metrics = observation.get
print(f"Input row count: {metrics['input_row_count']}")
print(f"Error row count: {metrics['error_row_count']}")
print(f"Warning row count: {metrics['warning_row_count']}")
print(f"Valid row count: {metrics['valid_row_count']}")
print(f"Check metrics: {metrics['check_metrics']}")  # per-check error/warning counts as JSON

Writing Metrics to a Table

End-to-end engine methods for applying checks (e.g. apply_checks_and_save_in_table, apply_checks_by_metadata_and_save_in_table, save_results_in_table, apply_checks_and_save_in_tables, apply_checks_and_save_in_tables_for_patterns) can write summary metrics into a table automatically using configuration. Metrics can be written to a table in batch or streaming. You can write metrics for different datasets or workloads into a common metrics table to track data quality over time centrally. For batch inputs, these methods can write summary metrics without writing output or quarantine tables by providing metrics_config only.

The name specified in the DQMetricsObserver is recorded as the run_name column in the metrics table. It is recommended to assign a unique name to the observer for each job or table to enable efficient filtering when centralizing metrics in a single table.

Compute requirements

Writing metrics directly to a results table is not supported for classic compute clusters in Standard access mode with Databricks runtime versions earlier than 17.3LTS.

Writing Metrics to a Table in Batch

Summary metrics can be written to a table when calling DQEngine methods to apply checks and write output data, quarantine data, or metrics only. When the input data is read as a batch source, metrics will be collected and written in batch.

Python

from databricks.labs.dqx import check_funcs
from databricks.labs.dqx.engine import DQEngine
from databricks.labs.dqx.metrics_observer import DQMetricsObserver
from databricks.labs.dqx.rule import DQRowRule, DQDatasetRule
from databricks.labs.dqx.config import InputConfig, OutputConfig
from databricks.sdk import WorkspaceClient

# Define the checks
checks = [
  DQRowRule(
    criticality="warn",
    check_func=check_funcs.is_not_null,
    column="col3",
  ),
  DQDatasetRule(
    criticality="error",
    check_func=check_funcs.is_unique,
    columns=["col1", "col2"],
  ),
  DQRowRule(
    name="email_invalid_format",
    criticality="error",
    check_func=check_funcs.regex_match,
    column="email",
    check_func_kwargs={"regex": r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"},
  ),
]

# Create the observer
observer = DQMetricsObserver(name="dq_metrics")

# Create the engine with the metrics observer
engine = DQEngine(WorkspaceClient(), observer=observer)

# Create the input config for a batch data source
input_config = InputConfig("main.demo.input_data")

# Create the output, quarantine, and metrics configs
output_config = OutputConfig("main.demo.valid_data")
quarantine_config = OutputConfig("main.demo.quarantine_data")  # optional
metrics_config = OutputConfig("main.demo.metrics_data")  # optional

# Option 1: Apply checks and save metrics
valid_df, quarantine_df, observation = engine.apply_checks_and_split(df, checks)
quarantine_df.count()  # Trigger an action to populate metrics (e.g. count, save to a table), otherwise accessing them will result in a stall
engine.save_summary_metrics(
  observed_metrics=observation.get,
  metrics_config=metrics_config,
  input_config=input_config,  # used as info only
  output_config=output_config,  # used as info only
  quarantine_config=quarantine_config,  # used as info only
  checks_location="checks.yml",  # used as info only
)

# Option 2: Use End to End method: read the data, apply the checks, write data to valid and quarantine tables, and write metrics to the metrics table
# By default, checks are applied to the entire input table. See next section for incremental support with Streaming.
engine.apply_checks_and_save_in_table(
    checks=checks,  # or provide checks_location and run_config_name to auto-load from checks storage
    input_config=input_config,
    output_config=output_config,
    quarantine_config=quarantine_config,
    metrics_config=metrics_config
)

# Option 3: Use End to End method to write summary metrics only for a batch input
engine.apply_checks_and_save_in_table(
    checks=checks,  # or provide checks_location and run_config_name to auto-load from checks storage
    input_config=input_config,
    metrics_config=metrics_config
)

Writing Metrics to a Table with Streaming

Summary metrics can also be written in streaming. When the input data is read as a streaming source, metrics will be written for each streaming micro-batch:

Supported methods

Metrics are not directly accessible from the returned Spark Observation when data is processed with streaming. You must use streaming metrics listener or end-to-end methods that persist the output in tables after quality checks are applied (e.g. e.g. apply_checks_and_save_in_table, apply_checks_by_metadata_and_save_in_table, save_results_in_table, apply_checks_and_save_in_tables, apply_checks_and_save_in_tables_for_patterns). Metrics-only streaming writes are not supported because DQX needs an output or quarantine streaming query to emit observed metrics.

Python

import time
from databricks.labs.dqx import check_funcs
from databricks.labs.dqx.engine import DQEngine
from databricks.labs.dqx.metrics_observer import DQMetricsObserver
from databricks.labs.dqx.rule import DQRowRule, DQDatasetRule
from databricks.labs.dqx.config import InputConfig, OutputConfig
from databricks.sdk import WorkspaceClient

# Define the checks
checks = [
  DQRowRule(
    criticality="warn",
    check_func=check_funcs.is_not_null,
    column="col3",
  ),
  DQRowRule(
    name="email_invalid_format",
    criticality="error",
    check_func=check_funcs.regex_match,
    column="email",
    check_func_kwargs={"regex": r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"},
  ),
]

# Create the observer
observer = DQMetricsObserver(name="dq_metrics")

# Create the engine with the metrics observer
engine = DQEngine(WorkspaceClient(), observer=observer)

# Create the input config for a streaming data source
input_config = InputConfig("main.demo.input_data", is_streaming=True)

# Create the output, quarantine, and metrics configs
output_config = OutputConfig(
  location="main.demo.valid_data",
  trigger={"availableNow": True},  # stop the stream once all data is processed
  options={"checkpointLocation": "/tmp/checkpoint/valid_data"}  # use a Volume in production for persistence
)
quarantine_config = OutputConfig(
  location="main.demo.quarantine_data",
  trigger={"availableNow": True},  # stop the stream once all data is processed
  options={"checkpointLocation": "/tmp/checkpoint/quarantine_data"}  # use a Volume in production for persistence
)
metrics_config = OutputConfig("main.demo.metrics_data")  # streaming configuration not required for metrics

# Option 1: Apply checks and save metrics
df = spark.readStream.table(input_config.location)
valid_df, quarantine_df, observation = engine.apply_checks_and_split(df, checks)
output_query = valid_df.writeStream.format(output_config.format).outputMode(output_config.mode).options(**output_config.options).trigger(**output_config.trigger).toTable(output_config.location)
quarantine_query = quarantine_df.writeStream.format(quarantine_config.format).outputMode(quarantine_config.mode).options(**quarantine_config.options).trigger(**quarantine_config.trigger).toTable(quarantine_config.location)

listener = engine.get_streaming_metrics_listener(
  input_config=input_config,
  output_config=output_config,
  quarantine_config=quarantine_config,
  metrics_config=metrics_config,
  target_query_id=quarantine_query.id,
)
# for streaming writing metrics requires a stream listener, observation cannot be accessed directly
# this adds a global listener for the current Spark session so do not add it again if reusing the same session
spark.streams.addListener(listener)

output_query.awaitTermination()
quarantine_query.awaitTermination()

# Option 2: Use End-to-End method: read the data, apply the checks, write data to valid and quarantine tables, and write metrics to the metrics table
# Output and quarantine data will be written in streaming and summary metrics will be written for each micro-batch
engine.apply_checks_and_save_in_table(
    checks=checks,  # or provide checks_location and run_config_name to auto-load from checks storage
    input_config=input_config,
    output_config=output_config,
    quarantine_config=quarantine_config,
    metrics_config=metrics_config
)

Saving Results and Metrics to a Table

Summary metrics can also be written to a table when calling save_results_in_table. After applying checks, pass the Spark Observation and output DataFrame(s) with the appropriate output configuration. For batch results, you can pass only the Spark Observation and metrics_config to write summary metrics without writing row-level output. This is supported for both batch and streaming.

Python

from databricks.labs.dqx import check_funcs
from databricks.labs.dqx.engine import DQEngine
from databricks.labs.dqx.metrics_observer import DQMetricsObserver
from databricks.labs.dqx.rule import DQRowRule, DQDatasetRule
from databricks.labs.dqx.config import OutputConfig
from databricks.sdk import WorkspaceClient

# Define the checks
checks = [
  DQRowRule(
    criticality="warn",
    check_func=check_funcs.is_not_null,
    column="col3",
  ),
  DQDatasetRule(
    criticality="error",
    check_func=check_funcs.is_unique,
    columns=["col1", "col2"],
  ),
  DQRowRule(
    name="email_invalid_format",
    criticality="error",
    check_func=check_funcs.regex_match,
    column="email",
    check_func_kwargs={"regex": r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"},
  ),
]

# Create the observer
observer = DQMetricsObserver(name="dq_metrics")

# Create the engine with the metrics observer
engine = DQEngine(WorkspaceClient(), observer=observer)

# Apply checks, split and get metrics
valid_df, quarantine_df, observation = engine.apply_checks_and_split(df, checks)

# Apply checks and get metrics
#checked_df, observation = engine.apply_checks(df, checks)

# Create the output, quarantine, and metrics configs
output_config = OutputConfig("main.demo.valid_data")
quarantine_config = OutputConfig("main.demo.quarantine_data")  # optional
metrics_config = OutputConfig("main.demo.metrics_data")  # optional

# Write the data to valid and quarantine tables, and write metrics to the metrics table
engine.save_results_in_table(
    output_df=valid_df,
    quarantine_df=quarantine_df,
    observation=observation,
    output_config=output_config,
    quarantine_config=quarantine_config,
    metrics_config=metrics_config
)

# Write only summary metrics for batch results
checked_df, observation = engine.apply_checks(df, checks)
checked_df.count()  # Trigger the observation before saving metrics
engine.save_results_in_table(
    observation=observation,
    metrics_config=metrics_config
)

Summary Metrics in Spark Declarative Pipelines (Lakeflow / DLT)

Available since v0.16.0

To compute summary metrics in Spark Declarative Pipelines (a.k.a. SDP, Lakeflow Pipelines, or DLT), use compute_summary_metrics, passing the same checks you applied to the input data. For information about why the DQObserver cannot be used, see Why not the observer.

Use compute_summary_metrics on batch data

compute_summary_metrics computes aggregates over the entire input dataset. It cannot be used with append-mode streaming queries that write data using df.writeStream without a watermark. Use a materialized view, batch write, or foreachBatch sink instead.

Strategies for writing summary metrics

There are several approaches for computing and persisting summary metrics in a pipeline. Choose the approach based on your pipeline type, whether you need a snapshot or a history, and whether multiple pipelines must write into a single, shared metrics table:

Option	Metrics shape	Pipeline type	Shared table across pipelines	Use when
Materialized view	Cumulative snapshot over the whole table	Batch or streaming	No — pipeline-managed dataset, owned by one pipeline	You want current totals and the simplest setup, don't need per-run history, and don't need to consolidate metrics across multiple pipelines.
foreachBatch sink	Per-batch history (appended)	Streaming	Yes — appends to an external Delta table	You want a history of metrics, better performance on large or growing tables (aggregates only the current micro-batch, not the whole table), or to centralize metrics from multiple pipelines into one table.
Windowed streaming table	Per-time-window history (appended)	Streaming	No — pipeline-managed dataset, owned by one pipeline	You want time-windowed metrics and have a timestamp column and watermark to window on.
Batch companion job	Per-run history (appended)	Batch	Yes — appends to an external Delta table	You want the simplest approach for keeping a history for 1 or more batch pipelines in a single metrics table.

Choosing a history strategy

Pick one history strategy per metrics table. Differet strategies require different columns and have outputs that cannot be appended into the same table (i.e. the windowed streaming table carries a window column plus metric columns while the foreachBatch-sink and batch-companion outputs use the long OBSERVATION_TABLE_SCHEMA shape).

Pipeline-managed options (e.g. materialized views or windowed streaming tables) are defined by a single query in a single pipeline and cannot share a target metrics table. To centralize metrics from several checked tables in one pipeline, unionByName the results of compute_summary_metrics inside a single materialized view. To centralize metrics across pipelines, use a foreachBatch sink or a batch companion job. Set a distinct observer name and input_config for each source so rows stay filterable.

Using a materialized view (cumulative snapshot)

Persist the checked data as a table so the result columns are available downstream, then add a materialized view that calls compute_summary_metrics over that table. After each pipeline update, the view contains a cumulative snapshot over all rows in the checked table. Metrics are cumulative totals representing the current quality of the entire dataset and do not contain any per-run history.

Python

import dlt
from databricks.labs.dqx.engine import DQEngine
from databricks.labs.dqx.metrics_observer import DQMetricsObserver
from databricks.sdk import WorkspaceClient

# compute_summary_metrics requires an observer on the engine; it reads any custom_metrics from it.
engine = DQEngine(WorkspaceClient(), observer=DQMetricsObserver())

@dlt.view
def bronze():
  return spark.readStream.table("catalog.schema.input")

# 1. Apply checks and persist the checked data as a table.
@dlt.table
def silver():
  df = dlt.read_stream("bronze")
  return engine.apply_checks_by_metadata(df, checks)

# 2. Compute summary metrics as a materialized view over the checked table.
@dlt.table
def dq_summary_metrics():
  df = dlt.read("silver")
  return engine.compute_summary_metrics(df, checks=checks)

See the Lakeflow pipeline demo for a complete example, or the quarantine variant that splits valid and invalid records into separate tables.

Incremental materialized views

compute_summary_metrics stamps each metrics row with a run_id and run_time. Without static values these are non-deterministic across pipeline runs. To enable incremental updates to the metrics, override the run_time_overwrite and run_id_overwrite with static values when creating the DQEngine by passing ExtraParams.

Using a foreachBatch sink (per-batch history)

Compute metrics per micro-batch using a foreachBatch sink. The sink function receives each streaming micro-batch, applies quality checks, writes the checked rows, and appends the per-batch compute_summary_metrics output. Use this when:

• You want a history of metrics for each streaming micro-batch rather than a cumulative snapshot
• You need better performance on large or growing tables; Only the current batch is aggregated rather than the whole table
• You want to share a metrics table across pipelines; The sink appends to an external Delta table (not a pipeline-managed dataset); Several pipelines can write into a common metrics table; Set a distinct observer name per source so rows stay filterable

Python

from pyspark import pipelines as dp

@dp.table
def bronze():
  return spark.readStream.table("catalog.schema.input")

# A foreachBatch sink writes to tables *outside* the pipeline, so use fully-qualified names —
# unqualified names would be captured as pipeline-managed streaming tables and rejected.
@dp.foreach_batch_sink(name="silver_sink")
def silver_sink(batch_df, batch_id):
  checked_df = engine.apply_checks_by_metadata(batch_df, checks)
  checked_df.write.format("delta").mode("append").saveAsTable("catalog.schema.silver")
  
  metrics_df = engine.compute_summary_metrics(checked_df, checks=checks)
  metrics_df.write.format("delta").mode("append").saveAsTable("catalog.schema.dq_summary_metrics")

@dp.append_flow(target="silver_sink")
def silver_flow():
  return spark.readStream.table("bronze")

See the Lakeflow foreachBatch sink demo and its quarantine variant for complete examples.

Using a windowed streaming table (per-window history)

If you prefer time-windowed metrics, add a watermark on a timestamp column and aggregate the observer's metric expressions per time window. Each window's metrics are appended once the window completes:

Python

import pyspark.sql.functions as F
from databricks.labs.dqx.metrics_observer import DQMetricsObserver

# pass check names to get_metrics(check_names) to also include the per-check breakdown
observer = DQMetricsObserver()

@dlt.table
def dq_summary_metrics():
  df = dlt.read_stream("silver").withWatermark("event_time", "10 minutes")
  metric_exprs = [F.expr(m) for m in observer.get_metrics()]
  return df.groupBy(F.window("event_time", "1 hour")).agg(*metric_exprs)

Watermark requirements for time-windowed metrics

Time-windowed metrics require that the source data contains a timestamp column to window on and a watermark. Ingestion or event time are commonly used. Each window's metrics are emitted only once the window closes and the watermark passes a configured delay interval. Late rows are dropped and not considered in the computed metrics. The output carries a window column plus the metric columns.

Using a batch companion job (per-run history)

For a batch pipeline, compute the metrics in a companion job or scheduled task and append the compute_summary_metrics output. This creates a row-set per run, stamped with a distinct run_id and run_time.

Python

metrics_df = engine.compute_summary_metrics(spark.read.table("catalog.schema.silver"), checks=checks)
metrics_df.write.mode("append").saveAsTable("catalog.schema.dq_metrics")

Why not the observer directly?

In Spark Declarative Pipelines, the pipeline runtime (not your code) triggers the write action. The Spark Observation that DQX's metrics observer relies on is never populated and the streaming listener receives no events. DQX automatically detects the pipeline runtime and automatically skips observe(), so apply_checks* returns the checked DataFrame without observable metrics. compute_summary_metrics computes the same metrics as an aggregation and returns a lazily-evaluated DataFrame with the same schema as the observer path. Pass the same checks to get the full per-check breakdown (including checks with zero violations). The metrics can be centralized alongside other batch and streaming workloads.

Metrics require a DQMetricsObserver

compute_summary_metrics uses the engine's DQMetricsObserver. An observer must be configured when initializing the engine. Calling compute_summary_metrics without an observer raises an InvalidParameterError.

Configuring Custom Metrics

Custom metrics are collected in addition to the built-in metrics. Pass custom metrics as Spark SQL expressions when creating the DQMetricsObserver. Custom metrics should be defined as Spark SQL expressions with column aliases and will be accessible by their alias.

Each custom metric expression must return a scalar aggregate value. metric_value is a string column, so a non-scalar result (e.g. an array or struct) is stringified opaquely and is hard to consume downstream — aggregate to a single scalar per metric.

Python

from databricks.labs.dqx.engine import DQEngine
from databricks.labs.dqx.metrics_observer import DQMetricsObserver
from databricks.labs.dqx.config import InputConfig, OutputConfig

# Define custom metrics
custom_metrics = [
  "sum(array_size(_errors)) as total_check_errors",
  "sum(array_size(_warnings)) as total_check_warnings",
]

# Create the observer with custom metrics
observer = DQMetricsObserver(
  name="business_metrics",
  custom_metrics=custom_metrics
)

# Create the engine with the optional observer
engine = DQEngine(WorkspaceClient(), observer=observer)

# Apply checks and get metrics
checked_df, observation = engine.apply_checks_by_metadata(df, checks)

# Trigger an action to populate metrics (e.g., count, save to a table).
# Without triggering an action, metrics will not be populated, and accessing them will result in a stall.
checked_df.count()  # Example action to ensure metrics are computed

# Access metrics
metrics = observation.get
print(f"Input row count: {metrics['input_row_count']}")
print(f"Error row count: {metrics['error_row_count']}")
print(f"Warning row count: {metrics['warning_row_count']}")
print(f"Valid row count: {metrics['valid_row_count']}")
print(f"Check metrics: {metrics['check_metrics']}")  # per-check error/warning counts as JSON
print(f"Total check errors: {metrics['total_check_errors']}")
print(f"Total check warnings: {metrics['total_check_warnings']}")

Workflows Integration

No-Code Approach (Workflows)

When using DQX workflows, summary metrics are automatically configured based on your installation configuration file:

1. Installation Configuration: During installation, specify metrics table and custom metrics.
2. Automatic Observer Creation: Workflows automatically create DQMetricsObserver when metrics are configured.
3. Metrics Persistence: Metrics are automatically saved to the configured table after each workflow run. The run_name is set to 'dqx' by default.

Configuration File Example

Metrics can be defined in the metrics_config section of your configuration file.

run_configs:
- name: production
  input_config:
    location: main.raw.sales_data
    format: delta
  output_config:
    location: main.clean.sales_data
    format: delta
    mode: append
  quarantine_config:
    location: main.quarantine.sales_data
    format: delta
    mode: append
  metrics_config:  # Summary metrics configuration
    location: main.analytics.dq_metrics
    format: delta
    mode: append
  checks_location: main.config.quality_checks
  
# Global custom metrics (applied to all run configs)
custom_metrics:
  - "avg(amount) as average_transaction_amount"
  - "sum(case when region = 'US' then amount else 0 end) as us_revenue"
  - "count(distinct customer_id) as unique_customers"

Once the config is defined you can start the workflows as usual using Databricks CLI, Databricks UI, Databricks API, or via Workflows scheduling:

# Run quality checker workflow with metrics enabled using Databricks CLI
databricks labs dqx apply-checks --run-config "production"

# Run end-to-end workflow with metrics enabled using Databricks CLI
databricks labs dqx e2e --run-config "production"

Metrics Table Schema

Summary metrics can be written and centralized in a delta table. The metrics table contains the following fields:

Column Name	Column Type	Description
run_id	STRING	Unique run ID recorded in the summary metrics as well as detailed quality checking results to enable cross-referencing. When reusing the same DQEngine and observer instances, the run ID stays the same. Each apply checks execution does not generate a new run ID for the same instance. It is only changed when new engine and observer (if using one) is created.
run_name	STRING	Name of the metrics observer: name passed to DQMetricsObserver, or 'dqx' when applying checks using Workflows.
input_location	STRING	Location of the input dataset (table name or file path), if known.
output_location	STRING	Location of the output dataset (table name or file path).
quarantine_location	STRING	Location of the quarantine dataset (table name or file path), if used.
checks_location	STRING	Location where checks are stored (table name or file path), if known.
rule_set_fingerprint	STRING	SHA-256 fingerprint of the rule set used for this run. Enables correlation with checks storage and results. Populated automatically when applying checks or saving results.
metric_name	STRING	Name of the metric (e.g., 'input_row_count').
metric_value	STRING	Value of the metric. All values are stored as strings — numeric metrics (e.g. input_row_count) are string-encoded integers, and check_metrics is a JSON string: [{"check_name": "...", "error_count": N, "warning_count": N}, ...]. Cast to the appropriate type when querying (e.g. CAST(metric_value AS INT) for counts, or parse JSON for check_metrics).
run_time	TIMESTAMP	Run timestamp when the summary metrics were calculated.
error_column_name	STRING	Name of the error column in the output or quarantine table containing per row quality checking results (default: '_errors').
warning_column_name	STRING	Name of the warning column in the output or quarantine table containing per row quality checking results (default: '_warnings').
user_metadata	MAP[STRING, STRING]	User-defined, run-level metadata.

Tracking from the summary table

DQX tracks rules automatically and records run_id, checks_location, and rule_set_fingerprint in the summary metrics table. You can use these fields to trace from aggregate metrics to details: start from a summary row, use run_id (and optionally rule_set_fingerprint) to filter _errors and _warnings in the output or quarantine tables and see which rows failed which checks, then use checks_location and rule_set_fingerprint to load the exact rule set version from checks storage and confirm which rules were applied.

Best Practices

Performance Considerations

1. Batch Metrics Collection: Collect metrics during regular data processing after output is written.
2. Monitor Metrics Overhead: Complex custom metrics may impact processing performance.

Monitoring and Alerting using metrics table

Use cases:

1. Track Trends: Monitor metrics over time to identify data quality degradation.
2. Set Thresholds: Establish acceptable ranges for error rates and warning counts.
3. Alert on Anomalies: Set up alerts when metrics deviate significantly from historical patterns, e.g. by using Databricks SQL Alerts.

The example below shows how you can analyze metrics persisted to a table:

SQL

/* EXAMPLE: Identify quality degradation */
WITH daily_metrics AS (
  SELECT
    date_trunc('day', run_time) as run_date,
    input_location,
    metric_name,
    CAST(metric_value AS DOUBLE) as metric_value
  FROM
    main.analytics.dq_metrics
  WHERE
    run_time >= current_date - INTERVAL 30 DAYS
    AND metric_name IN ('input_row_count', 'error_row_count', 'warning_row_count')
),
pivoted_metrics AS (
  SELECT
    run_date,
    input_location,
    MAX(CASE WHEN metric_name = 'input_row_count' THEN metric_value END) as input_count,
    MAX(CASE WHEN metric_name = 'error_row_count' THEN metric_value END) as error_count,
    MAX(CASE WHEN metric_name = 'warning_row_count' THEN metric_value END) as warning_count
  FROM daily_metrics
  GROUP BY run_date, input_location
)
SELECT
  run_date,
  input_location,
  avg(error_count * 100.0 / NULLIF(input_count, 0)) as avg_error_rate,
  avg(warning_count * 100.0 / NULLIF(input_count, 0)) as avg_warning_rate
FROM
  pivoted_metrics
WHERE
  input_count > 0
GROUP BY
  run_date, input_location
ORDER BY
  run_date DESC, input_location

When you need to explore detailed row-level quality results based on summary metrics for troubleshooting, the example below illustrates how to do so:

Python

import pyspark.sql.functions as F

# fetch any metrics row as an example
metrics_row = spark.table(metrics_table_name).collect()[0]
run_id = metrics_row["run_id"]
output_table_name = metrics_row["output_location"]

# retrieve detailed results
output_df = spark.table(output_table_name)

# extract errors
results_df = output_df.select(
  F.explode(F.col("_errors")).alias("result"),
).select(F.expr("result.*"))

# extract warnings
results_df = output_df.select(
  F.explode(F.col("_warnings")).alias("result"),
).select(F.expr("result.*"))

# Fetch detailed quality results using the run_id from summary metrics
filtered_results_df = results_df.filter(F.col("run_id") == run_id)  # filter, or join
filtered_results_df.show()