Data Quality Checks in Airflow: SQL Check Operators Explained

Airflow runs data quality checks through the SQL check operators in the common-sql provider. You add them as tasks in a DAG, they execute against a connection you have already configured, and a failed check fails the task. There are six of them, they behave differently enough that picking the wrong one wastes an afternoon, and two of them have defaults that will quietly let bad data through. This is how each one works, with runnable examples and the behavior that is not obvious from the class name.

For the side by side of every route to Airflow data quality with its documented ceiling, including where third party frameworks fit, see our Airflow data quality checks page. This article is the practical walkthrough.

Install the provider first

The operators live in apache-airflow-providers-common-sql, not in core Airflow. Most managed distributions bundle it, but if you are building your own image it is an explicit dependency:

pip install apache-airflow-providers-common-sql

Every operator below takes a conn_id pointing at a connection you have already defined, and an optional database that overrides whatever the connection specifies.

SQLColumnCheckOperator: per column assertions

This is where most teams start, and it is the right default for column properties. You describe the columns you care about in a column_mapping dictionary, and each column gets one or more checks:

from airflow.providers.common.sql.operators.sql import SQLColumnCheckOperator

check_orders_columns = SQLColumnCheckOperator(
    task_id="check_orders_columns",
    conn_id="snowflake_prod",
    table="analytics.public.orders",
    column_mapping={
        "order_id": {
            "null_check": {"equal_to": 0},
            "unique_check": {"equal_to": 0},
        },
        "amount": {
            "min": {"geq_to": 0},
            "max": {"leq_to": 1000000},
            "null_check": {"equal_to": 0, "tolerance": 0.01},
        },
        "region": {
            "distinct_check": {"leq_to": 6},
        },
    },
)

The available checks are null_check, distinct_check, unique_check, min and max. The conditions are greater_than, geq_to, less_than, leq_to and equal_to. Three details are worth knowing before you write a hundred of these.

equal_to is not compatible with other conditions. You cannot ask for a value that equals something and is also bounded by something else in the same check. Pick one.

tolerance is a percentage, not a count. It is documented as a percentage that the result may be out of bounds but still considered successful, which is useful for a null check on a column where a handful of nulls is normal and a thousand is not.

accept_none is true by default, and it converts None values returned by the query to zeros. This is the one that bites. If an upstream change empties a column entirely, an aggregate like min returns None, the operator reads it as zero, and a condition of geq_to: 0 passes on a column with no data in it. If a null aggregate should be a failure rather than a number, set it explicitly:

check_amount = SQLColumnCheckOperator(
    task_id="check_amount",
    conn_id="snowflake_prod",
    table="analytics.public.orders",
    accept_none=False,
    column_mapping={"amount": {"min": {"geq_to": 0}}},
)

On a large table, add a partition_clause so you are not scanning history on every run. It can be set at operator, column or check level, which means you can scan only today's partition for the cheap checks and widen it for the ones that need more rows.

SQLTableCheckOperator: conditions across the whole table

When the thing you want to assert is not a property of one column, this is the general purpose tool. Each entry in checks carries a check_statement containing SQL that resolves to a boolean:

from airflow.providers.common.sql.operators.sql import SQLTableCheckOperator

check_orders_table = SQLTableCheckOperator(
    task_id="check_orders_table",
    conn_id="snowflake_prod",
    table="analytics.public.orders",
    checks={
        "row_count_check": {"check_statement": "COUNT(*) > 1000"},
        "total_matches_lines": {
            "check_statement": "SUM(amount) = SUM(line_total)"
        },
        "no_future_dates": {
            "check_statement": "MAX(ordered_at) <= CURRENT_TIMESTAMP()"
        },
    },
)

Cross column arithmetic is the case this handles well and nothing else does. Reconciling a header total against the sum of its lines is a genuinely valuable check and it cannot be expressed as a column property.

The trade off is that every threshold in there is a number somebody chose. COUNT(*) > 1000 is correct until the business has a quiet week or triples, and then it is either a false alarm or a missed incident. That is not a flaw in the operator, it is the nature of a static assertion, and it is the main reason volume checks tend to migrate out of the DAG over time.

SQLCheckOperator and the zero trap

SQLCheckOperator is the oldest and least forgiving. It runs a query that returns a single row and evaluates every value in that row using Python's bool() casting, failing if any value is falsy. The documented falsy set includes False, 0, an empty string, an empty list and an empty dictionary.

Read that again with a counting query in mind, because this is the classic bug:

# WRONG: returns 0 when the data is clean, and 0 is falsy,
# so this check fails exactly when everything is fine.
SQLCheckOperator(
    task_id="check_no_orphans",
    conn_id="snowflake_prod",
    sql="SELECT COUNT(*) FROM orders WHERE customer_id IS NULL",
)

Write it as a comparison that returns a boolean instead:

# RIGHT
SQLCheckOperator(
    task_id="check_no_orphans",
    conn_id="snowflake_prod",
    sql="""
        SELECT COUNT(*) = 0 AS no_orphans
        FROM orders
        WHERE customer_id IS NULL
    """,
)

You can return several columns in the row and all of them must be truthy for the task to pass, which makes it a compact way to bundle related assertions when you do not want six task boxes in the graph view.

SQLIntervalCheckOperator: comparing against last week

This is the only native operator that compares a metric against history rather than a constant, which makes it the closest thing Airflow has to anomaly detection. You supply metrics_thresholds, a dictionary mapping each metric to an acceptable ratio:

from airflow.providers.common.sql.operators.sql import SQLIntervalCheckOperator

check_volume_drift = SQLIntervalCheckOperator(
    task_id="check_volume_drift",
    conn_id="snowflake_prod",
    table="analytics.public.orders",
    date_filter_column="order_date",
    days_back=-7,
    metrics_thresholds={
        "COUNT(*)": 1.5,
        "SUM(amount)": 1.5,
    },
)

The defaults tell you what it assumes about your tables. days_back defaults to -7, comparing against the same day last week, which is a good choice because it holds day of week constant. date_filter_column defaults to ds, the conventional partition column name, so any table that does not use that name needs the parameter set, as above. ratio_formula defaults to max_over_min, which divides the larger value by the smaller and therefore treats a doubling and a halving identically; use relative_diff when you want a signed proportional difference. ignore_zero defaults to True, skipping the comparison when a value is zero rather than producing an infinite ratio.

It is a competent week over week drift check for a daily partitioned fact table. What it is not is a baseline: it knows one prior data point, not a distribution, so it cannot tell you that today is unusual against the last ninety days, and it cannot learn that the first of the month is always triple. Choosing the ratio is the whole problem, and the operator hands that problem back to you.

SQLValueCheckOperator and SQLThresholdCheckOperator

Both compare a single query result. SQLValueCheckOperator takes a pass_value and an optional numerical tolerance. SQLThresholdCheckOperator takes a min_threshold and max_threshold, and usefully, either of those can be a query rather than a number, which lets you bound a metric by something derived from another table:

from airflow.providers.common.sql.operators.sql import SQLThresholdCheckOperator

check_daily_revenue = SQLThresholdCheckOperator(
    task_id="check_daily_revenue",
    conn_id="snowflake_prod",
    sql="SELECT SUM(amount) FROM orders WHERE order_date = CURRENT_DATE()",
    min_threshold="SELECT 0.5 * AVG(daily_total) FROM daily_revenue_history",
    max_threshold="SELECT 2.0 * AVG(daily_total) FROM daily_revenue_history",
)

That pattern, thresholds derived from a history table you maintain yourself, is how teams approximate a baseline inside Airflow. It works, and it means you now own a history table, its retention and its backfill.

Where to put the checks in the DAG

Two placements, and the difference matters more than the operator choice. Put checks between the load and the publish when you can stage data first, so a failed check stops bad rows from reaching anything downstream. Put them after the write when staging is impractical, accepting that a failure means the bad data is already visible and you are now doing cleanup.

The staged version is worth the extra table:

load_to_staging >> check_staging >> swap_into_prod >> notify

Also set retries=0 on check tasks. Retrying a data quality check just runs the same assertion against the same data and delays the alert, unless the check is genuinely reading something that is still being written, in which case a sensor is the right tool rather than a retry.

Three limits to plan around

Results are not stored as data. A check passes or fails, the detail lands in the task log, and the task instance records the outcome. That is enough to debug this morning and not enough to answer any question containing the word trend. How often has this table been late this quarter, is the null rate drifting, what is the normal row count on a Tuesday. Astronomer names this directly as the reason to reach for a third party framework: you use one when you want to collect the results of your data quality checks in a central place.

Checks do not run when the DAG does not. This is the structural one. Astronomer's own comparison states that DAG level checks cannot check data when there is a problem with Airflow, while platform level checks monitor data even if there is a problem with Airflow. Rank the ways a table actually goes wrong in production and the list starts with the load not running at all: a paused DAG nobody unpaused, a scheduler outage, a sensor that timed out and skipped everything downstream, a task removed in a refactor, expired credentials. In every one of those cases no check failed, because no check ran, and an empty alert channel looks identical whether the pipeline is healthy or never started. Pairing data checks with plain infrastructure uptime monitoring on the scheduler and the APIs feeding it covers part of that gap, and the rest needs something watching the tables on its own schedule.

Coverage is the count of tasks somebody wrote. column_mapping only ever looks at the columns it names, and checks only evaluates the statements it contains. New table, new task, new pull request, new review. The tables that break are rarely the instrumented ones, they are the ones added eighteen months ago by a team that has since reorganized. Adding a check requires DAG code changes and deployment, so checks accumulate where somebody is already paying attention and the long tail stays uncovered.

Do I still need SQL check operators if I have data observability?

Yes, and you should keep them. The operators are good at exactly what a learned baseline cannot know: that a discount percentage never exceeds 40, that a status column only holds one of five values, that a header total reconciles against its lines, that a foreign key resolves. Those are business rules, they belong next to the transformation that produces them, and they are cheap to express. What they should stop carrying is outage detection, volume anomalies, schema drift and freshness, all of which need history, baselines and a run that does not depend on the scheduler. Splitting it that way is also what keeps a DAG readable instead of growing a check task per column until the pipeline itself is hard to see.

If you want the tables your DAGs write watched continuously, including on the mornings the DAG never fired, data pipeline monitoring covers freshness, volume, schema and distribution across Snowflake, BigQuery, Databricks and Redshift, and Airflow monitoring shows how it sits alongside an existing deployment. Pricing is published, starting at 99 dollars a month, with a 14 day trial that needs no card.