Запуск робочих навантажень SQL

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

Volker Janz

Senior Developer Advocate at Astronomer

Чому SQL в Airflow?

Робочі навантаження SQL — найтиповіший випадок для Airflow
Нехай база даних виконує важку роботу
Airflow керує коли і де
База даних відповідає за як

Оркестрація SQL в Airflow

SQLExecuteQueryOperator

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


SQLExecuteQueryOperator(
    task_id="load_sales",
    conn_id="duckdb_analytics",
    sql="INSERT INTO sales SELECT * FROM staging",
)

Працює з будь-якою базою даних, що має сумісний провайдер Airflow
PostgreSQL, Snowflake, BigQuery, DuckDB та інші

Підключення

Зберігайте облікові дані поза кодом
Кожне підключення має ID, тип, хост, порт, логін
Створення різними способами: UI, CLI, API або змінні середовища
Посилайтесь на них через conn_id в операторах

Інтерфейс підключень Airflow

Побудова SQL-конвеєра

from airflow.sdk import dag, task
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator

@dag(schedule="@daily")
def sales_pipeline():

    aggregate = SQLExecuteQueryOperator(
        task_id="aggregate_daily_sales",
        conn_id="duckdb_analytics",
        sql="""
            INSERT INTO daily_summary (order_date, total_orders, total_revenue)
            SELECT order_date, COUNT(*), SUM(revenue)
            FROM raw_orders GROUP BY order_date
        """,
    )

Зовнішні SQL-файли

@dag(
    schedule="@daily",
    template_searchpath="/path/to/include/sql",
)
def sales_pipeline():
    aggregate = SQLExecuteQueryOperator(
        task_id="aggregate_daily_sales",
        conn_id="duckdb_analytics",
        sql="aggregate_sales.sql",
    )

Встановіть template_searchpath у @dag
У полі sql вкажіть назву файла
Зберігайте скрипти й бізнес-логіку за межами папки dags/ 💡

Структура файлів проєкту

Шаблони Jinja в SQL-файлах

DELETE FROM daily_summary WHERE order_date = '{{ ds }}';

INSERT INTO daily_summary (order_date, total_orders, total_revenue)
SELECT order_date, COUNT(*), SUM(revenue)
FROM raw_orders
WHERE order_date = '{{ ds }}'
GROUP BY order_date;

{{ ds }} підставляє логічну дату (YYYY-MM-DD)
DELETE, потім INSERT: ідемпотентний шаблон з Розділу 2
Повторний запуск для тієї ж дати дає той самий результат

params vs parameters

params (рендеринг Jinja)

SQLExecuteQueryOperator(
    sql="""SELECT * FROM orders
           WHERE product = '{{ params.product }}'""",
    params={"product": user_input},
)

Значення інтерполюється в рядок SQL
Уразливе до SQL-інʼєкцій

parameters (звʼязування на рівні БД)

SQLExecuteQueryOperator(
    sql="SELECT * FROM orders
         WHERE product = $product",
    parameters={"product": user_input},
)

Значення передається драйверу бази даних
Безпечно від інʼєкцій: драйвер обробляє екранування

Від локально до продакшну з Astro

Astronomer's Astro: керована платформа Airflow

Astro CLI: локальний Airflow однією командою

Розробляйте локально, безшовно розгортайте в продакшн

Astro CLI та платформа

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

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