on_failure_callback 在动态 dag 中

Question

我正在从列表中动态创建 dag，并想向其中一项任务添加 on_failure_callback。我试过下面的代码，但回调似乎没有被执行。

dag_ids = ['dag_a', 'dag_b', 'dag_c']

for dag_id in dag_ids:

    def failure_callback():
        logging.info('Inside failure callback for {}'.format(dag_id))

    def python_callable(dag_id):
        logging.info('Inside python callable for {}'.format(dag_id))
        raise Exception('Exception raised for dag_id {}'.format(dag_id))

    yesterday = datetime.datetime.combine(
        datetime.datetime.today() - datetime.timedelta(1),
        datetime.datetime.min.time())

    default_args = {
        'start_date': yesterday
    }    

    dag = models.DAG(
            dag_id,
            schedule_interval=None,
            catchup=False,
            default_args=default_args)

    with dag:

        python_task = PythonOperator(
            task_id='python_task',
            python_callable=python_callable,
            op_kwargs={'dag_id': dag_id},
            on_failure_callback=failure_callback,
            dag=dag)

        python_task

    globals()[dag_id] = dag

知道我做错了什么吗？

编辑：

根据建议，我将 dag_id 传递给失败回调。但是 airflow 不是 dag_id，而是传递上下文字典。除了上下文字典之外，关于如何将额外参数传递给失败回调有什么想法吗？

ERROR - Inside failure callback for {u'next_execution_date': None, u'dag_run': <DagRun dag_a @ 2019-02-19 19:23:54.006241: manual__2019-02-19T19:23:54.006241, externally triggered: True>, u'tomorrow_ds_nodash': u'20190220', u'run_id': 'manual__2019-02-19T19:23:54.006241', u'test_mode': False, u'prev_execution_date': None, u'conf': <module 'airflow.configuration' from '/usr/local/lib/airflow/airflow/configuration.py'>, u'tables': None, u'task_instance_key_str': u'dag_a__python_task__20190219', u'END_DATE': '2019-02-19', u'execution_date': datetime.datetime(2019, 2, 19, 19, 23, 54, 6241), u'ts': '2019-02-19T19:23:54.006241', u'macros': <module 'airflow.macros' from '/usr/local/lib/airflow/airflow/macros/__init__.py'>, u'params': {}, u'ti': <TaskInstance: dag_a.python_task 2019-02-19 19:23:54.006241 [failed]>, u'var': {u'json': None, u'value': None}, u'ds_nodash': u'20190219', u'dag': <DAG: dag_a>, u'end_date': '2019-02-19', u'latest_date': '2019-02-19', u'ds': '2019-02-19', u'task_instance': <TaskInstance: dag_a.python_task 2019-02-19 19:23:54.006241 [failed]>, u'yesterday_ds_nodash': u'20190218', u'task': <Task(PythonOperator): python_task>, u'yesterday_ds': '2019-02-18', u'ts_nodash': u'20190219T192354.006241', u'tomorrow_ds': '2019-02-20'}

参考了问题 here 并且成功了！

Answer 1

在 for 循环中，只需将 dag_id 传递给 failure_callback，然后您应该能够看到来自 failure_callback() 的日志。

def failure_callback(dag_id):
    logging.info('Inside failure callback for {}'.format(dag_id))

Answer 2

使用 partials 包让它工作。更新后的代码如下。

from functools import partial

dag_ids = ['dag_a', 'dag_b', 'dag_c']

for dag_id in dag_ids:

    def failure_callback(dag_id, context):
        logging.error('Inside failure callback for {}'.format(dag_id))

    def python_callable(dag_id):
        logging.error('Inside python callable for {}'.format(dag_id))
        raise Exception('Exception raised for dag_id {}'.format(dag_id))

    yesterday = datetime.datetime.combine(
        datetime.datetime.today() - datetime.timedelta(1),
        datetime.datetime.min.time())

    default_args = {
        'start_date': yesterday
    }    

    dag = models.DAG(
            dag_id,
            # Continue to run DAG once per day
            schedule_interval=None,
            catchup=False,
            default_args=default_args)

    with dag:

        python_task = PythonOperator(
            task_id='python_task',
            python_callable=python_callable,
            op_kwargs={'dag_id': dag_id},
            on_failure_callback=partial(failure_callback, dag_id),
            dag=dag)

        python_task

    globals()[dag_id] = dag

on_failure_callback 在动态 dag 中

on_failure_callback in dynamic dags

python

airflow

google-cloud-composer