气流：如何将变量从 BaseOperator 执行的 python 脚本传递到 DAG 中的 on_failure_callback arg

如何解决气流：如何将变量从 BaseOperator 执行的 python 脚本传递到 DAG 中的 on_failure_callback arg

我有一个简单的 DAG，其中有一项需要 dockerized 的任务。基于其中包含 variable1 的脚本的运行，我需要将其提供给在 DAG 文件中创建的回调函数。如何解决这个问题？

with DAG(
    dag_id=DagName,default_args=default_args,schedule_interval='12 * * * *',on_failure_callback=callback_function(variable1),) as dag:

first_task = BaSEOperator(
    name='first-task',image=image,cmd=f'python script1.py {arg1}'.
)

解决方法

XComs 可用于传递变量。基本上，XComs 通过传递消息来帮助任务之间的交流。

您的任务会将值推送到 xcom，on_failure_callback 函数将从 xcom 中提取值。

这是一个简单的例子（为了简单起见，使用 PythonOperator）：

def on_failure(context):
    logger = logging.getLogger()
    logger.info("############## on_failure_callback ##############")
    my_variable = context["ti"].xcom_pull(task_ids="",key="my_variable")
    logger.info(my_variable)


def do_something(**context):
    logger = logging.getLogger()
    logger.info("############## do_something_func ##############")
    context["ti"].xcom_push(key="my_variable",value="hello")
    raise Exception("make the task fail")


with DAG(dag_id="example_dag",start_date=days_ago(2),schedule_interval='0 */1 * * *',on_failure_callback=on_failure) as dag:
    t1 = PythonOperator(task_id="task_1",python_callable=do_something,provide_context=True)