Я использую Airflow для выполнения задач KubernetSpodoperator , которые запускают изображение приложения Java.
Приложение Java Использование Beam Dataflow.
my -пользовательские операторы воздушного потока наследуют kubernetepodoperator . on_failure_callback Метод воздушного потока.
Но я получаю Airflowexception вместо реального исключения.
Я не могу добавить попробовать свой код Java, так как я использую Beam и Pipeline.Run () или Pipeline.WaitUntilfiniLfinis Блок Try-Catch.
Результат. Метод waituntilfinish () в трубопроводе пучка только отражает исключения, которые происходят во время установки задания. Он не отражает исключения времени выполнения, которые происходят при обработке данных.def task_failure(context):
exception = context.get('exception')
if exception:
exception_type = type(exception).__name__
default_args = {
'on_failure_callback': task_failure
}
with models.DAG(
dag_name,
default_args=default_args,
start_date=start_date,
schedule_interval=interval
) as dag:
< /code>
версии: < /p>
воздушный поток 2.2.5 < /li>
Композитор 2.1.9 < /li>
Kubernetes 1.29 < /li>
Java 17 < /li>
< /ul>
Подробнее здесь: https://stackoverflow.com/questions/787 ... lows-on-fa
