У меня DAG с воздушным потоком, выполняющий несколько задач BashOperator.Эти задачи в свою очередь вызывают API, который отправляет задания в кластеры EMR.API возвращает другой код завершения в зависимости от того, успешно ли выполнено задание EMR (0), не выполнено (1) или истекло ли время ожидания (124).Только поток задач с истекшим сроком должен быть повторен воздушным потоком.
Возможно ли иметь функцию обратного вызова при ошибке, которая будет захватывать код выхода и повторно запускать задачу?Если нет, то может ли это быть обработано некоторыми существующими функциями, которые я пропустил?
== Редактировать 1 == Произошло ли какое-то копание в источнике воздушного потока и обнаружилось это:
if sp.returncode:
raise AirflowException("Bash command failed")
Это указывает на то, чтоесли код выхода не равен 0, воздушный поток помечает задачу как невыполненную для всех других кодов выхода.По сути, для любого кода выхода, отличного от 0, воздушный поток будет повторять задачу на основе настроенного значения повторения.
Кроме того, если я добавлю логику в обратном вызове on_error для повторной попытки на основе кода ошибки, у меня не будетконтроль количества повторных попыток выполнения задачи.