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

Question

最初使用 dag callback ( on_failure_callback 和 on_success_callback )时，我认为它会在 success 完成时触发 fail 或 dag 状态(正如它在 dag 中定义的那样)。
但是它似乎在每个 task instance 而不是 dag run 处被实例化，因此如果 DAG 有 N 个任务，它将触发这些回调 N 次。

我正在 try catch 任务 ID，因此发送到 slack。阅读另一个 related question 我想出了以下内容:

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/docs/apache-airflow/stable/macros-ref.html