Перевірки якості даних

Створення конвеєрів даних з Airflow

Volker Janz

Senior Developer Advocate at Astronomer

Чому важливі «ворота» якості

Конвеєр без перевірок — це конвеєр, якому не можна довіряти
Одна зміна вище за течією спричиняє null або від'ємні значення
Усі наступні дашборди показують неправильні числа
«Воротa» якості ловлять проблеми на джерелі

Погані дані течуть униз за потоком

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: рахує null-значення (використайте equal_to: 0, щоб не було null)
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

Спершу завантажте, потім перевіряйте
Якщо перевірка падає, конвеєр зупиняється
Споживачі нижче за течією не побачать поганих даних

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

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}},
    },
)

Чи є null?
Чи в межах допустимого діапазону?

SQLTableCheckOperator

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

Чи має таблиця рядки?
Чи загальне значення в очікуваних межах?

Масштабний моніторинг якості з Astro

Astronomer's Astro Observe допомагає відстежувати стан конвеєрів
Відстеження SLA з попереджаннями до дедлайнів
Походження конвеєра між DAG і таблицями
Моніторинг якості даних у великому масштабі
Аналіз першопричин на основі ШІ

Лінійність у Astro Observe

Давайте потренуємось!

Створення конвеєрів даних з Airflow