Шаблони, ідемпотентність і бекфілінг

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

Volker Janz

Senior Developer Advocate at Astronomer

Шаблони Jinja в Airflow

@task.bash
def export_data():
    return "cp /data/sales.csv /export/sales_{{ ds }}.csv"

{{ ds }} підставляє логічну дату у форматі YYYY-MM-DD
Кожен запуск має свою дату, тож один і той самий таск дає різний результат на кожен запуск
Працює лише в полях із шаблонами (як-от bash_command, sql)

Проблема ідемпотентності

Запуск 1 (31 березня):

INSERT продажі за 31 березня
Результат: 3 рядки

Запуск 2 (повтор 31 березня):

INSERT продажі за 31 березня знову
Результат: 6 рядків (дублі!)

Дублікати рядків при повторному запуску

Шаблон «видалити, потім вставити»

staged_rows = SQLExecuteQueryOperator(
    task_id="get_staged_rows",
    conn_id="duckdb_default",
    sql="SELECT * FROM staging WHERE date = '{{ ds }}'",
)

load = SQLInsertRowsOperator(
    task_id="load_sales",
    conn_id="duckdb_default",
    table_name="sales",
    columns=["date", "product", "amount"],
    preoperator="DELETE FROM sales WHERE date = '{{ ds }}';",
    rows=staged_rows.output,
)

Також можна використати підходи UPSERT або MERGE

Видалити, потім вставити

Бекфілінг

Перепроцесіть діапазон історичних дат
Airflow створює один Dag run на логічну дату
Разом з ідемпотентністю бекфілінг є безпечним

Бекфілінг залежить від розкладу

Бекфілінг в інтерфейсі

Trigger DAG, оберіть Backfill, задайте дати From і To
Можна обрати Missing Runs, Missing and Errored Runs або All Runs
Задайте паралельність через Max Active Runs і змініть порядок виконання
Натисніть Run Backfill, щоб почати створення запусків

Інтерфейс бекфілінгу в Airflow

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

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