气流-从dag上下文回调中解析任务ID

Question

刚开始使用dag callback（on_failure_callback和on_success_callback）时，我认为完成后会触发successor fail状态dag（如dag中定义）。但是随后似乎每个实例都task instance没有实例化它dag run，因此，如果DAG具有N个任务，它将触发N次这些回调。

我正在尝试捕获任务ID，因此发送到松弛状态。在阅读另一个相关问题时，我想到了以下内容：

def success_msg(context):
    slack.slack_message(context['task_instance']); #send task-id to slack

def failure_msg(context):
    slack.slack_message(context['task_instance']); #send task-id to slack

default_args = {
    [...]
    'on_failure_callback': failure_msg,
    'on_success_callback': success_msg,
    [...]
}

但是它失败了，我应该如何解析上下文变量并因此获得任务ID？

Answer 1

您可以从上下文中使用任务对象访问任务。

context['task']应该是执行此操作的适当方法。要获取任务名称，请使用task_id：

context['task'].task_id

要查找上下文中可用的更多对象，您可以在此处浏览列表：https : //airflow.apache.org/macros.html