airflow on_failure_callback回调函数

william 收录于 Tools

2023-05-20 约 500 字预计阅读 1 分钟

警告

本文最后更新于 2023-05-20，文中内容可能已过时。

注册回调函数，一旦 DAG 触发相关条件，会执行该回调函数

支持 DAG 与 Operator

可以在 DAG 或者单独一个 Operator(Task) 里面定义回调，入参使用为 on_failure_callback=[callback_func_1, callback_func_2]

Callback Types

There are five types of task events that can trigger a callback:

Name	Description
`on_success_callback`	Invoked when the task succeeds
`on_failure_callback`	Invoked when the task fails
`sla_miss_callback`	Invoked when a task misses its defined SLA
`on_retry_callback`	Invoked when the task is up for retry
`on_execute_callback`	Invoked right before the task begins executing.

同时，在 airflow2.6.0 以上版本，还可以支持多个回调函数，将其放入一个 list 即可

As of Airflow 2.6.0, callbacks now supports a list of callback functions, allowing users to specify multiple functions to be executed in the desired event. Simply pass a list of callback functions to the callback args when defining your DAG/task callbacks: e.g on_failure_callback=[callback_func_1, callback_func_2]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64


from docker.types import Mount
from airflow import DAG
from airflow.utils.dates import days_ago
from airflow.decorators import task
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator
from airflow.providers.docker.operators.docker import DockerOperator
from wepy.utils.init import *

import pendulum
local_tz = pendulum.timezone("Asia/Shanghai")
DEFAULT_ARGS = {
    'owner'            : 'william',
    'email'            : 'xxx@xxx.com',
    'email_on_failure' : False,
    'email_on_retry'   : False,
    'depends_on_past'  : False,
    'retries'          : 3,
    'retry_delay'      : timedelta(minutes=1)
}

def dag_failure_alert(context):
    msg = f"""DAG has failed,
        ---------------------
        {context}
        ---------------------
        {context.get('dag')=}
        {context.get('dag_run')=}
        ---------------------
        {context.get('task')=}
        {context.get('task_instance')=}
        """
    log.err(msg)
    wx_test.send(msg, 'error')

def task_failure_alert(context):
    msg = f"""Task has failed,
        task_instance_key_str: {context['task_instance_key_str']}
        """
    log.err(msg)
    wx_test.send(msg, 'error')

def checking():
    if not cal.is_today_trading():
        raise Exception("Not TradingDay.")

with DAG(
    dag_id         = "hello",
    default_args   = DEFAULT_ARGS,
    schedule       = "*/1 * * *",
    start_date     = datetime(2023, 1, 1, tzinfo=local_tz),
    catchup        = False,
    dagrun_timeout = timedelta(seconds=60*3),
    tags           = ['hello'],
    on_failure_callback = [dag_failure_alert],
) as dag:
    checking = PythonOperator(
        task_id = 'checking',
        python_callable = checking,
        dag = dag,
        on_failure_callback = [task_failure_alert],
    )

    checking

airflow on_failure_callback回调函数

支持 DAG 与 Operator

Callback Types

相关内容