Can we pass x_com variable to the next task in DAG as a parameter?

user_15174345 0
Viewed 300

Please consider this scenario:

I have a cloud Dag on cloud composer that triggers a cloud function. That function hits an api and then stores tables in GCS. Now my Airflow DAG(using Cloud Composer) triggers next stage i.e a Dataproc job that takes the table from GCS and pushes into BQ but when I trigger my Dataproc workflow template, I am passing a parameter that is the name of the table from the dag itself and that parameter I want to pick from x_com.

Here is a code snippet throws an error that it is not defined

dataproc_job = dataproc_operator.DataprocWorkflowTemplateInstantiateOperator(
# The task id of your job
task_id="dataproc_job",
# The template id of your workflow
template_id="newwf1",
project_id='#######',
region="us-central1",
parameters={"TABLE_NAME":ti.xcom_pull(task_ids=simple_http}
)

How can I resolve this error and pass x_com value as a parameter to my next step in DAG ?

google-cloud-composer airflow google-cloud-platform
1 Answers

Assuming the parameters parameter can be templated (aka listed as a templated_field in the DataprocWorkflowTemplateInstantiateOperator), you can use Jinja expression to access the XCom value.

DataprocWorkflowTemplateInstantiateOperator(
    # The task id of your job
    task_id="dataproc_job",
    # The template id of your workflow
    template_id="newwf1",
    project_id='#######',
    region="us-central1", 
    parameters={"TABLE_NAME": "{{ ti.xcom_pull(task_ids='simple_http' }}"} 
)

More on Jinja templating in Airflow here as well as using XComs with templates mentioned in this doc.

As an aside, that operator looks like a very old Airflow 1 operator. If you are able to, I would highly recommend upgrading to Airflow 2. There is a myriad of both functional and performance improvements to make your Airflow and pipeline execution experience much better.

user_157723770
Related