气流：当 on_success_callback 执行带参数的函数时答案

【问题标题】：Airflow: when on_success_callback execute a function with parameters气流：当 on_success_callback 执行带参数的函数时
【发布时间】：2017-11-18 06:33:35
【问题描述】：

我想用我从任务传递的一个参数来执行一个函数。

这是我的带有状态参数的函数：

def sns_notify(state):
    client = boto3.client('sns')
    if state == "failed":
        message = config.get('sns', 'message') + state
    else:
        message = config.get('sns', 'message') + state
    response = client.publish(TargetArn=config.get('sns', 'target_arn'),
                              Message=message,
                              Subject=config.get('sns', 'subject'))
    return response

这是我的任务，状态为参数：

t1 = DummyOperator(task_id='Dummy-1', trigger_rule=TriggerRule.ALL_SUCCESS,
                   on_success_callback=sns_notify("ok"), dag=dag)

t2 = DummyOperator(task_id='Dummy-2', trigger_rule=TriggerRule.ONE_FAILED,
                   on_success_callback=sns_notify("failed"), dag=dag)

当我运行 dag 时，函数不会停止发送邮件（例如）

【问题讨论】：

response 行应该与else 我想。在这里加4个空格就大功告成了
这是另外一回事，我只想用 if,else 初始化变量 message。
不停止发送邮件是什么意思？在你杀死执行者之前，你会一直收到电子邮件吗？

标签： python scheduled-tasks scheduler airflow apache-airflow

【解决方案1】：

每次 DAG 被气流加载时，它都会执行sns_notify("ok")，因为您正在调用该函数。您只需要传递函数指针sns_notify，它将接收context。参见文档：https://airflow.apache.org/code.html

trigger_rule 与依赖任务的执行方式有关，因此与on_success_callback 无关。

我不确定如何将变量传递给这个回调 - 来这里寻找答案！

【讨论】：

我猜，您需要将参数添加到 context 字典中，on_success_callback callable 将其作为参数接收。

【解决方案2】：

Hoju 指出了确切的错误。

您可以使用函数式编程来帮助解决这个问题。

from functools import partial
send_success_notification = partial(sns_notify, "OK")
t1 = DummyOperator(task_id='Dummy-1', trigger_rule=TriggerRule.ALL_SUCCESS,
                   on_success_callback=send_success_notification , dag=dag)

send_failure_notification = partial(sns_notify, "FAILED")
t2 = DummyOperator(task_id='Dummy-2', trigger_rule=TriggerRule.ONE_FAILED,
                   on_success_callback=send_failure_notification, dag=dag)

【讨论】：