异常是如何传递到on_failure_callback？

在Airflow中，当任务失败时可以配置一个回调函数on_failure_callback来处理异常。异常可以通过两种方式传递到on_failure_callback：

1. 通过Python的try-except语句捕获异常，然后在异常处理代码块中调用on_failure_callback函数，将异常作为参数传递给该函数。

例如，在任务中定义了一个on_failure_callback函数：

def on_failure_callback(context):
    print("Task failed: {}".format(context['task_instance'].task_id))
    # do something to handle the exception, e.g. send an email

然后，在任务中可以使用try-except语句来捕获异常，并调用on_failure_callback函数：

def my_task():
    try:
        # do something
    except Exception as e:
        on_failure_callback(context)

2. 使用XCom将异常传递到on_failure_callback函数中。

例如，在任务中将异常传递到XCom：

def my_task():
    try:
        # do something
    except Exception as e:
        # set an XCom key-value pair to store the exception
        context['ti'].xcom_push(key='my_task_exception', value=str(e))

然后，在on_failure_callback函数中获取XCom值：

def on_failure_callback(context):
    exception = context['ti'].xcom_pull(key='my_task_exception')
    if exception:
        print("Task failed with exception: {}".format(exception))
        # do something to handle the exception, e.g. send an email