python - Airflow :当 on_success_callback 执行带参数的函数时

Question

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

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

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

这是我以 state 为参数的任务:

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 时，该函数不会停止发送邮件(例如)

Answer 1

每次通过 Airflow 加载 DAG 时，它都会执行 sns_notify("ok")因为您正在调用该函数。您只需要传递函数指针 sns_notify ，将收到 context .请参阅文档:https://airflow.apache.org/code.html
trigger_rule与依赖任务的执行方式有关，因此与 on_success_callback 无关.

但是，我不确定如何将变量传递给此回调 - 来到这里寻找答案!