DQX , A Spark-based framework for Data Quality Checks-Part 1

This is the first post in a multi-part series where we explore DQX , from basic setup to advanced configuration.

What is DQX?

DQX (Data Quality eXtended) is a data quality framework from Databricks Labs, built for Apache Spark. It lets you define, track, and handle data quality issues directly in your Python-based data pipelines.

• It’s a Python library available on PyPI, designed to catch data quality problems during processing, not after the data is already written
• It works natively with PySpark DataFrames and supports Lakeflow Pipelines (DLT) integration
• Invalid data can be dropped, flagged, or quarantined depending on your needs. Quality rules can be defined at both row and column levels, either in code or through configuration files (YAML/JSON)
• It can profile your input data automatically and suggest data quality rules based on what it finds
• Unlike traditional approaches where bad records land in your tables first and checks run after the fact, DQX validates data before it gets written, for both batch and streaming pipelines
• DQX is an open-source project meant for exploration. It is not officially supported by Databricks with SLAs , it’s provided as-is with no guarantees

Data Quality Check Types

Before jumping into code, it helps to understand how DQX is designed at a high level:

• DQX works by annotating rows in a DataFrame or table.
• After running checks, you get either two separate DataFrames (valid and quarantined) or a single annotated DataFrame with error and warning columns added to each row
• DQX is built with Spark performance in mind. Throughout the official documentation, you’ll find optimization notes on how checks are evaluated efficiently on Spark. As an example:

DQX gives you two ways to define checks: through YAML configuration files or directly in Python using DQX classes. Both approaches support the same features; It comes down to whether you prefer config-driven or code-driven pipelines.

Checks can be categorized in two ways:

Row-level vs. Dataset-level:

• Row-level checks validate each row individually. Think of things like checking for null values, verifying that a value falls within a given range, or matching a pattern
• Dataset-level checks require aggregation across rows. For example, checking whether an entire column is null, or making sure the most recent ingestion_timestamp in a DataFrame is not older than N days

Built-in vs. Custom:

• Built-in checks are ready-to-use functions that come with the DQX library, such as is_not_null, is_in_range, and others
• Custom checks are for cases where your validation logic isn’t covered by the built-ins. You write your own check function and plug it into DQX

A simple built-in rule could be written like:

# Ensure only 1 country code exists per country
- criticality: error
  check:
    function: is_aggr_not_greater_than
    arguments:
      column: country_code
      aggr_type: count_distinct
      group_by:
      - country
      limit: 1

A more complex custom check can be written like:

- criticality: error
  check:
    function: sql_query
    arguments:
      merge_columns:
        - sensor_id
      condition_column: condition
      msg: one of the sensor reading is greater than limit
      name: sensor_reading_check
      negate: false
      input_placeholder: sensor
      query: |
        WITH joined AS (
          SELECT
            sensor.*,
            COALESCE(specs.min_threshold, 100) AS effective_threshold
          FROM {{ sensor }} sensor
          LEFT JOIN {{ sensor_specs }} specs
            ON sensor.sensor_id = specs.sensor_id
        )
        SELECT
          sensor_id,
          MAX(CASE WHEN reading_value > effective_threshold THEN 1 ELSE 0 END) = 1 AS condition
        FROM joined
        GROUP BY sensor_id

For custom checks, we can either use:

• Python functions: define a custom logic, register it as a rule, and reuse it in YAML files or DQX classes
• SQL query: define a custom SQL query and use it as your logic. It’s not reusable like Python functions. Use it for more ad-hoc checks.

Criticality

Each check in DQX has a criticality level that controls what happens when a row fails. There are two options:

• warn: The row is flagged in the _warnings column, but it still passes through as valid. It will show up in the valid DataFrame
• error: The row is flagged in the _errors column and gets moved to the quarantined DataFrame. It won't make it to the output

This gives you the flexibility to separate hard failures from soft issues. For example, a missing customer_id might be an error worth quarantining, while an age value slightly out of range might just deserve a warning.

A Simple Example

Let’s put it all together with a minimal setup. We’ll define two built-in checks in YAML and run them on a PySpark DataFrame.

checks.yml file:

- criticality: error
  check:
    function: is_not_null_and_not_empty
    arguments:
      column: customer_id

- criticality: warn
  check:
    function: is_in_range
    arguments:
      column: age
      min_limit: 0
      max_limit: 120

running the checks:

from databricks.labs.dqx.engine import DQEngine
from databricks.labs.dqx.config import FileChecksStorageConfig
from databricks.sdk import WorkspaceClient

# Initialize the engine
dq_engine = DQEngine(WorkspaceClient())

# Load checks from YAML
checks = dq_engine.load_checks(
    config=FileChecksStorageConfig(location="checks.yml")
)

# Apply checks and split results
df_valid, df_quarantine = dq_engine.apply_checks_by_metadata_and_split(
    df, checks
)

That’s it. After running this, df_valid contains all rows that passed (including rows with warnings in the _warnings column), and df_quarantine contains the rows that failed an error-level check.

In the next post, I’ll dive deeper into the row-level vs. dataset-level checks.

References:

• DQX GitHub Documentation: https://databrickslabs.github.io/dqx/