这可能是由于默认情况下,Airflow任务中的on_failure_callback只会接收一个参数-异常信息。如果要在on_failure_callback中执行多行代码,则需要将代码包装在try-except块中,并将异常信息作为参数传递给它。例如:
def my_failure_callback(context):
try:
# 执行多行代码
pass
except Exception as e:
# 打印异常信息
print(str(e))
# 在DAG中定义on_failure_callback
dag = DAG(
'my_dag',
default_args=default_args,
on_failure_callback=my_failure_callback
)
请注意,on_failure_callback只有在任务失败时才会被执行,因此请确保你的代码中有处理任务失败情况的逻辑。