Проверки качества данных

Построение пайплайнов данных с Airflow

Volker Janz

Senior Developer Advocate at Astronomer

Зачем нужен контроль качества

Пайплайн без проверок — это пайплайн, которому нельзя доверять
Одно изменение выше по цепочке порождает пустые значения или отрицательные числа
Все последующие дашборды показывают неверные данные
Контроль качества выявляет проблемы в источнике

Некачественные данные распространяются вниз по цепочке

SQLColumnCheckOperator

from airflow.providers.common.sql.operators.sql import
    (SQLExecuteQueryOperator, SQLColumnCheckOperator)


check_columns = SQLColumnCheckOperator(
    task_id="check_columns",
    conn_id="duckdb_analytics",
    table="daily_summary",
    column_mapping={
        "total_revenue": {"min": {"greater_than": 0}},
        "total_orders": {"null_check": {"equal_to": 0}},
    },
)

Типы проверок столбцов

column_mapping={
    "total_revenue": {"min": {"greater_than": 0}},
    "total_orders": {"null_check": {"equal_to": 0}},
    "order_date": {"distinct_check": {"greater_than": 0}},
}

min: проверяет минимальное значение в столбце
null_check: считает пустые значения (условие equal_to: 0 означает отсутствие nulls)
distinct_check: считает уникальные значения
Доступные проверки: null_check, unique_check, distinct_check, min, max
Доступные операторы сравнения: equal_to, greater_than, geq_to, less_than, leq_to

SQLTableCheckOperator

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


check_table = SQLTableCheckOperator(
    task_id="check_table",
    conn_id="duckdb_analytics",
    table="daily_summary",
    checks={
        "row_count_check": {
            "check_statement": "COUNT(*) > 0",
        },
    },
)

Проверяет свойства всей таблицы
check_statement: любое SQL-выражение, возвращающее true или false

Цепочка проверок качества

@dag(schedule="@daily", template_searchpath="/path/to/sql")
def sales_pipeline():

    load = SQLExecuteQueryOperator(
        task_id="load_daily_sales",
        conn_id="duckdb_analytics",
        sql="aggregate_sales.sql",
    )

    check_columns = SQLColumnCheckOperator(...)
    check_table = SQLTableCheckOperator(...)

    load >> check_columns >> check_table

Сначала загрузка, затем валидация
При сбое любой проверки пайплайн останавливается
Потребители данных никогда не увидят некорректные данные

Проверки столбцов и таблиц

SQLColumnCheckOperator

check_columns = SQLColumnCheckOperator(
    task_id="check_columns",
    conn_id="duckdb_analytics",
    table="daily_summary",
    column_mapping={
        "total_revenue": {"min": {"greater_than": 0}},
        "total_orders": {"null_check": {"equal_to": 0}},
    },
)

Есть ли пустые значения?
Находятся ли значения в допустимом диапазоне?

SQLTableCheckOperator

check_table = SQLTableCheckOperator(
    task_id="check_table",
    conn_id="duckdb_analytics",
    table="daily_summary",
    checks={
        "row_count_check": {
            "check_statement": "COUNT(*) > 0",
        },
    },
)

Есть ли в таблице строки?
Соответствует ли итог ожидаемым границам?

Мониторинг качества в масштабе с Astro

Astro Observe от Astronomer помогает отслеживать состояние пайплайнов
Контроль SLA с оповещениями до истечения срока
Линейность пайплайнов на уровне DAG и таблиц
Мониторинг качества данных в масштабе
Анализ первопричин на основе ИИ

Линейность в Astro Observe

Давайте потренируемся!

Построение пайплайнов данных с Airflow