python之Airflow之处理 DAG 回调的正确方法

我有一个 DAG，然后每当它成功或失败时，我希望它触发一个发布到 Slack 的方法。

我的 DAG args 如下所示:

default_args = { 
    [...] 
    'on_failure_callback': slack.slack_message(sad_message), 
    'on_success_callback': slack.slack_message(happy_message), 
    [...] 
} 

DAG 定义本身:

dag = DAG( 
    dag_id = dag_name_id, 
    default_args=default_args, 
    description='load data from mysql to S3', 
    schedule_interval='*/10 * * * *', 
    catchup=False 
      ) 

但是当我检查 Slack 时，每分钟有超过 100 条消息，就好像在每个调度程序心跳和每个日志上进行评估一样，它确实运行了成功和失败方法，就好像它对同一任务有效和无效一样实例(不好)。

我应该如何正确使用 on_failure_callback 和 on_success_callback 来处理 dags 状态并调用自定义方法？

请您参考如下方法:

它创建消息的原因是因为当您定义default_args 时，您正在执行函数。您只需要传递函数定义而不执行它。

由于函数有一个参数，它会变得有点棘手。您可以定义两个部分函数或定义两个包装函数。

所以你可以这样做:

from functools import partial 
 
success_msg = partial(slack.slack_message, happy_message); 
failure_msg = partial(slack.slack_message, sad_message); 
 
default_args = { 
    [...] 
    'on_failure_callback': failure_msg 
    'on_success_callback': success_msg 
    [...] 
} 

或

def success_msg(): 
    slack.slack_message(happy_message); 
 
def failure_msg(): 
    slack.slack_message(sad_message); 
 
default_args = { 
    [...] 
    'on_failure_callback': failure_msg 
    'on_success_callback': success_msg 
    [...] 
} 

在任一方法中，请注意函数定义 failure_msg 和 success_msg 是如何被传递的，而不是它们在执行时给出的结果。