在 Airflow 2.0 中使用 Taskflow API 传递争论 [英] Passing arguements using Taskflow API in Airflow 2.0

问题描述

我正在使用 REST API 将参数传递给基于任务流的 Dag.看看这个论坛上提出的类似问题，下面似乎是访问传递参数的常用方法.

I am passing arguments to a Taskflow based Dag using the REST API. Looking at similar issues raised on this forum, it seems below are the usual ways of accessing the passed arguments.

#From inside a template field or file:
{{ dag_run.conf['key'] }}
#Or when context is available, e.g. within a python callable of the PythonOperator:
context['dag_run'].conf['key'] 

我试图获取上下文字典

@dag(default_args=default_args, schedule_interval=None, start_date=days_ago(2), params=None)
def classic_xgb(**context):
    """
    ### TaskFlow API Tutorial Documentation
    [here](https://airflow.apache.org/docs/stable/tutorial_taskflow_api.html)
    """
    @task()
    def extract():
        print("context is ", context)

输出是 现在我如何获得作为输入参数传递给 Dag 的 conf 字典?我需要在我的 python 代码中使用参数，因此模板选项对我来说似乎不可行.

The output is <airflow.models.dagparam.DagParam object at 0x7f735c634510> Now how do I get to the conf dictionary that was passed as the input arguement to the Dag? I need to use the arguements in my python code and so templating option does not seem viable for me.

任何帮助将不胜感激.

谢谢

此致，

阿迪尔

推荐答案

有一个新功能 get_current_context() 在 Airflow 2.0 中获取上下文.获得上下文字典后，params"键包含通过 REST API 发送到 Dag 的参数.下面的代码解决了这个问题.

There is a new function get_current_context() to fetch the context in Airflow 2.0. Once you have the context dict, the 'params' key contains the arguments sent to the Dag via REST API. The following code solved the issue.

from airflow.operators.python import task, get_current_context

default_args = {
    'owner': 'airflow',
}
@dag(default_args=default_args, schedule_interval=None, start_date=days_ago(2))
def classic_xgb(**kwargs):

    """
    @task()
    def extract():
        context = get_current_context()

这篇关于在 Airflow 2.0 中使用 Taskflow API 传递争论的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！