Update example DAG files to latest syntax

Hello,

When we stop releasing backport packages, we will be able to update all example DAGs to use the new syntax for handling outputs introduced by AIP-31.

This is an example change.

 product_set_update = CloudVisionUpdateProductSetOperator(
     location=GCP_VISION_LOCATION,
-    product_set_id="{{ task_instance.xcom_pull('product_set_create') }}",
+    product_set_id=product_set_create.output,
     product_set=ProductSet(display_name='My Product Set 2'), 
    task_id='product_set_update',
)

Best regards,
Kamil Bregułła

Or Maybe we can do automated refactoring for that one ? Sounds doable.

@turbaszek

CC: @turbaszek

This should be probably part of #9041

Hey, I am new here :). Can I contribute on this issue as my first contribution? Has someone already addressed this issue?

@hsnprsd

@hsnprsd feel free to take this one 🚀 I'm happy to help with eventual questions. I think, as a good start we should refactor airflow/example_dags/* to use .output instead of jinja whenever it is possible (those DAGs are not part of providers).

@mik-laj

@mik-laj @potiuk @turbaszek Is this enhancement still desired? I could take a crack at it if it is.

Sure. Take a look please :)

@mik-laj

@mik-laj @potiuk @turbaszek Would you all also like to see PythonOperator use be converted the @task decorator or would that kind of update be considered out of scope for the enhancement?

I think this could be done as extra task afterwards

Going through these example DAGs there are a number of xcom_pull() uses in which values from the XCom are directly accessed via an index and/or key as well as a for loop in Jinja. Trying to rewrite these instances using the new syntax is awfully cumbersome IMO. For example:

# Access by dict key
job_id="{{task_instance.xcom_pull('start_python_job_dataflow_runner_async')['dataflow_job_id']}}"

# Access by index
object_name="{{ task_instance.xcom_pull('upload_sheet_to_gcs')[0] }}"

# Using a for loop and accessing by dict key
echo_cmd = """
{% for m in task_instance.xcom_pull('pull_messages') %}
    echo "AckID: {{ m.get('ackId') }}, Base64-Encoded: {{ m.get('message') }}"
{% endfor %}
"""

The output property of operators doesn't seem to elegantly handle this type of access currently. I can use a workaround with ugly hacks like these:

# Dict key access workaround (index access would be written similarly)
start_python_job_dataflow_runner_async_output = str(start_python_job_dataflow_runner_async.output).strip("{ }")

wait_for_python_job_dataflow_runner_async_done = DataflowJobStatusSensor(
        task_id="wait-for-python-job-async-done",
        job_id="{{{{ {start_python_job_dataflow_runner_async_output}['dataflow_job_id'] }}}}",
        expected_statuses={DataflowJobStatus.JOB_STATE_DONE},
        project_id=GCP_PROJECT_ID,
        location='us-central1',
    )

# For loop workaround
pull_message_output = str(pull_messages.output).strip("{ }")

echo_cmd = f"""
    {{% for m in {pull_messages_output} %}}
        echo "AckID: {{{{ m.get('ackId') }}}}, Base64-Encoded: {{{{ m.get('message') }}}}"
    {{% endfor %}}
    """

I logged #16618 as a feature request to enhance the use of XComArgs to handle such access. I was wondering if you all thought it was worth implementing these workarounds or to leave them as-is for now. Perhaps there is a cleaner way to mimic the same result between XComArg and xcom_pull()? What do you think?

@josh-fell

@josh-fell Can't you do start_python_job_dataflow_runner_async.output['dataflow_job_id'] or similar? Looking at the docs that looks like it should work.

Oh, no I'm mis-reading what/where the Key is used.

@turbaszek

@turbaszek @mik-laj @potiuk All of the PRs to update the example DAGs have now been merged. This issue can be closed if you feel they have satisfied the scope of this issue.

@josh-fell

@josh-fell thanks. Closed!

@josh-fell

Fantastic! Thanks @josh-fell !

Navigation Menu

Repository navigation

Update example DAG files to latest syntax #10285

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Update example DAG files to latest syntax #10285

Description

Activity

potiuk commented on Aug 11, 2020

potiuk commented on Aug 11, 2020

turbaszek commented on Aug 11, 2020

hsnprsd commented on Nov 14, 2020

turbaszek commented on Nov 14, 2020

josh-fell commented on Jun 21, 2021

potiuk commented on Jun 22, 2021

josh-fell commented on Jun 24, 2021

potiuk commented on Jun 24, 2021

josh-fell commented on Jun 29, 2021

ashb commented on Jun 29, 2021

josh-fell commented on Sep 2, 2021

mik-laj commented on Sep 2, 2021

potiuk commented on Sep 2, 2021

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions