Тригери та відмовостійкість

Вступ до Apache Airflow на Python

Mike Metzger

Data Engineer

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

Тригери — керують, коли можна запускати завдання
- Визначають завдання, що виконуються за певних умов
- Тригер за замовчуванням — all success

Ілюстрація: тригери керують часом запуску завдань

Відмовостійкість — дає змогу відновлюватися всередині завдання
- Повторний запуск окремого завдання

Ілюстрація: відмовостійкість повторно запускає збоєве завдання

Правила тригерів

Перед стартом завдання перевіряйте стан попередніх
Дозволяє визначити, чи мають завдання продовжуватися
За замовчуванням усі попередні мають успішно завершитись
Правила тригерів дозволяють це змінити

Ілюстрація: правила тригерів перевіряють стани попередніх завдань

Застосування правил тригерів

Завдання сповіщення
Завдання прибирання
Умовне виконання

Ілюстрація: застосування правил тригерів — сповіщення та прибирання

Ключові правила тригерів

all_success — усі попередні завдання успішні, без пропусків
all_failed — усі попередні завдання з помилкою
all_done — усі попередні завершені, незалежно від результату
one_failed — щонайменше одне завдання з помилкою
one_success — щонайменше одне завдання успішне
none_failed — усі попередні успішні або пропущені

Реалізація правил тригерів

from airflow.utils.trigger_rule import TriggerRule
Правила тригерів застосовують до декоратора @task
trigger_rule=TriggerRule.<TriggerRule enum>

@task(trigger_rule=TriggerRule.ALL_SUCCESS)
def run_if_everything_succeeds:
  print('All previous tasks succeeded!')

Атрибути відмовостійкості

retries — параметр завдання, що визначає, скільки разів Airflow має повторити збійне завдання перед позначенням як невдале
retry_delay — параметр завдання з timedelta, що задає паузу між спробами

@task(
  trigger_rule=TriggerRule.ONE_SUCCESS,
  retries=3,
  retry_delay=timedelta(minutes=5)
)

TriggerDagRunOperator

Дозволяє одному Dag запускати інший Dag
Надає код і Dags
from airflow.providers.standard.operators.trigger_dagrun import TriggerDagRunOperator
trigger_dag_id — має збігатися з dag_id
wait_for_completion — блокує інші завдання до завершення
poke_interval — як часто перевіряти
conf — дані для передачі дочірньому Dag

Приклад TriggerDagRun

trigger_child = TriggerDagRunOperator(
        task_id="trigger_child_pipeline",
        trigger_dag_id="child_pipeline_dag",   
        wait_for_completion=True,              
        poke_interval=30,                      
        conf={                                 
            "source": "s3://my-bucket/raw/",
        },
    )

task1 >> trigger_child    # Спочатку виконайте перше завдання, потім запустіть дочірній Dag
trigger_child >> cleanup  # Виконайте прибирання після завершення дочірнього Dag

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

Вступ до Apache Airflow на Python