Как сделать retry нескольких связанных задач в Apache Airflow?
Есть стандартная ETL задача, взять данные из источника и положить в PostgreSQL.
Для этого есть Apache Airflow (2.2.3) и скрипты на python.
Но есть нюансы:
- К полученным данным часто идут тяжелые запросы, поэтому используются материальные вьюхи, которые надо обновлять после сбора.
- Данные не всегда собираются в полном объеме, поэтому на тасках сбора стоит несколько retry, но материальные вьюхи надо обновлять после каждого (ну или почти каждого) retry у upstream таска.
В DAG сейчас это выглядит примерно так:
два таска: A - PythonOperator, который собирает данные и кладет их в базу, B - PostgresOperator, который делает запрос на refresh materialized view.
Связаны последовательно: A >> B
Проблема в том, что A имеет несколько retry, и пока они все не закончатся (успешно или нет это не так важно) таск B не запустится.
Можно было бы запихнуть логику рефреша в таск A, но тогда пропадает возможность сделать примерно следующее
[A0, A1, A2] >> B (делать рефреш после выполнения нескольких upstream тасков).
Идеально, мне кажется, было бы запихнуть это все в TaskGroup и делать retry на всю группу, но к сожалению, Airflow не позволяет делать retry на группу. Раньше был Subdag, который теперь deprecated и заменен на TaskGroup.
Простой
Средний
