airflow - 将其他参数传递给 on_failure_callback

Question

我想将其他参数传递给我的 on_failure_callback 函数，但它似乎只需要“上下文”。我如何将其他参数传递给该函数......特别是因为我想在一个单独的模块中定义该函数，以便它可以在我所有的 DAGS 中使用。

我当前的 default_args 如下所示:

default_args = {
  'owner': 'Me',
  'depends_on_past': True,
  'start_date': datetime(2016,01,01),
  'email': ['me@me.com'],
  'email_on_failure': False,
  'email_on_retry': False,
  'retries': 1,
  'retry_delay': timedelta(minutes=1),
  'on_failure_callback': notify_failure,
  'max_active_runs': 1
}

如果我尝试这样的 Airflow 提示:

default_args = {
  'owner': 'Me',
  'depends_on_past': True,
  'start_date': datetime(2016,01,01),
  'email': ['me@me.com'],
  'email_on_failure': False,
  'email_on_retry': False,
  'retries': 1,
  'retry_delay': timedelta(minutes=1),
  'on_failure_callback': notify_failure(context,arg1,arg2),
  'max_active_runs': 1
}

所以不知道如何将 arg1 和 arg2 传递给我想在一个单独的模块中定义的 notify_failure 函数，我可以简单地导入我的 DAG

Answer 1

假设 args 是您可以在 DAG 级别定义的内容，那么您可以使用 partials 包。 IE:

from functools import partial

def generic_failure(arg1, arg2, context):
  # do whatever

default_args = {
  'owner': 'Me',
  'depends_on_past': True,
  'start_date': datetime(2016,01,01),
  'email': ['me@me.com'],
  'email_on_failure': False,
  'email_on_retry': False,
  'retries': 1,
  'retry_delay': timedelta(minutes=1),
  'on_failure_callback': partial(generic_failure, arg1, arg2),
  'max_active_runs': 1
}

调用 partial(generic_failure, arg1, arg2)将返回一个函数，但是 generic_failure 中还剩下许多参数。 , 在上面的例子中只是单个参数 context