Запуск 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 и 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

Astro от Astronomer: управляемая платформа Airflow

Astro CLI: локальный Airflow одной командой

Разрабатывайте локально, развёртывайте в продакшен без лишних усилий

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

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

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