Updating example DAG files to latest syntax via Taskflow API #16837
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
- Converted `xcom_pull()` calls to use `XComArg` where applicable - Updated `context` access via `kwargs` to use `get_current_context` - Removed explicit task dependencies which are created implicitly via `XComArgs` - Added comments on the implicit `XComArg` task dependencies - Removed and refactored `default_args` code pattern
- Converted `xcom_pull()` calls to use `XComArg` where applicable - Updated `context` access via `kwargs` to use `get_current_context` - Removed explicit task dependencies which are created implicitly via `XComArgs` - Added comments on the implicit `XComArg` task dependencies - Removed and refactored `default_args` code pattern
…/josh-fell/airflow into taskflow-api-update-example-dags
boring-cyborg
bot
added
provider:cncf-kubernetes
Contributor
|
I think this PR should be broken down into smaller PRs. Reviewing 86 files is not going to be easy. |
Contributor
Author
Solid suggestion. I will close this PR and open smaller, more manageable ones. Thanks for the feedback! |
Merged
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Closes #10285
xcom_pull()calls use an operator's.outputproperty as well access ofTaskInstanceobjects from context to useget_current_context()functionXComArgsfor transparencydefault_argspattern where necessary as requested by @ashb (i.e. removed a separateddefault_argsdeclaration for deference for declaration as part of theDAGobject).outputrefactoringAn detailed summary of all changes made as part of this PR can be found below:
xcom_pull()?default_argspattern.default_argspattern.default_argspattern.default_argspattern.default_argspattern.default_argspattern.default_argspattern.default_argspattern.default_argspattern.default_argspattern.default_argspattern.default_argspattern.default_argspattern.default_argspattern.default_argspattern.Updated to use
get_current_context().get_current_context()based on previous PR comments to show new and old means ofXComaccess.Removed explicit task dependencies that are created via
XComArgs.Removed unneeded
default_argspattern.default_argspattern.default_argspattern.default_argspattern.XComArgs.Removed unneeded
default_argspattern.XComArgs.Refactored
default_argspattern.default_argspattern.XComArgs.Refactored
default_argspattern.XComArgobject does not support accessing specific values of an iterableXComvalue elegantly. Did not update this occurrence:"{{ task_instance.xcom_pull(task_ids='add_steps', key='return_value')[0] }}".All other
xcom_pull()calls have been updated.Removed explicit task dependencies that are created via
XComArgs.Refactored
default_argspattern.XComArgobject does not support accessing specific values of an iterableXComvalue. Could not update any occurrences in this DAG.XComArgs.Updated to use
get_current_context().XComArgobject does not support accessing specific values of an iterableXComvalue. Could not update any occurrences in this DAG.default_argspattern.default_argspattern.XComArgs.Removed unneeded
default_argspattern.default_argspattern.default_argspattern.default_argspattern.XComArgobject does not support accessing specific values of an iterableXComvalue elegantly. Did not update this occurrence:"echo \"{{ task_instance.xcom_pull('write-xcom')[0] }}\"".Removed unneeded
default_argspattern.XComArgobject does not support accessing specific values of an iterableXComvalue elegantly. Did not update this occurrence:"{{ task_instance.xcom_pull(task_ids='spark_pi_submit')['metadata']['name'] }}".Refactored
default_argspattern.default_argspattern.default_argspattern.XComArgs.Refactored
default_argspattern.Refactored
ShortCircuitOperatorcallable logic.default_argspattern.default_argspattern.XComArgs.XComArgs.XComArgs.XComArgobject does not support accessing specific values of an iterableXComvalue elegantly. Did not update the following occurrences:"{{ extract_object_id(task_instance.xcom_pull('list_tables_spec_task')[0]) }}"'{{ task_instance.xcom_pull("create_dataset_task")["name"] }}'"{{ get_target_column_spec(task_instance.xcom_pull('list_columns_spec_task'), target) }}"All other
xcom_pull()calls have been updated.Removed explicit task dependencies that are created via
XComArgs.XComArgs.XComArgs.XComArgs.XComArgs.XComArgs.default_argspattern.XComArgs.XComArgobject does not support accessing specific values of an iterableXComvalue elegantly. Did not update any occurrences in this DAG.XComArgobject does not support accessing specific values of an iterableXComvalue elegantly. Did not update any occurrences in this DAG.XComArgobject does not support accessing specific values of an iterableXComvalue elegantly. Did not update this occurrence:"user-{{ task_instance.xcom_pull('get-instance')['persistence_iam_identity']"All other
xcom_pull()calls have been updated.XComArgobject does not support accessing specific values of an iterableXComvalue elegantly. Did not update this occurrence:"echo \"{{ task_instance.xcom_pull('lookup_entry')['display_name'] }}\""All other
xcom_pull()calls have been updated.XComArgs.Removed duplicate/redundant task.
XComArgobject does not support accessing specific values of an iterableXComvalue elegantly. Did not update any occurrences in this DAG.Removed unused
default_argsvariable.XComArgobject does not support accessing specific values of an iterableXComvalue elegantly. Did not update any occurrences in this DAG.XComArgobject does not support accessing specific values of an iterableXComvalue elegantly. Did not update this occurrence:"echo \"{{ task_instance.xcom_pull('lookup_entry')['display_name'] }}\""All other
xcom_pull()calls have been updated.XComArgobject does not support accessing specific values of an iterableXComvalue elegantly. Did not update any occurrences in this DAG.XComArgs.XComArgobject does not support accessing specific values of an iterableXComvalue elegantly. Did not update these occurrences:"{{task_instance.xcom_pull('export_task')['response']['outputUrl'].split('/')[2] }}""{{ '/'.join(task_instance.xcom_pull('export_task')['response']['outputUrl'].split('/')[3:]) }}"All other
xcom_pull()calls have been updated.Removed explicit task dependencies that are created via
XComArgs.default_argsto DAG as there was logic to populate the dict but not being applied.XComArgobject does not support accessing specific values of an iterableXComvalue elegantly. Did not update any occurrences in this DAG.XComArgobject does not support accessing specific values of an iterableXComvalue elegantly. Did not update any occurrences in this DAG.default_argspattern.XComArgobject does not support accessing specific values of an iterableXComvalue elegantly. Did not update the following occurrence:"""{% for m in task_instance.xcom_pull('pull_messages') %} echo "AckID: {{ m.get('ackId') }}, Base64-Encoded: {{ m.get('message') }}" {% endfor %}"""All other
xcom_pull()calls have been updated.Removed explicit task dependencies that are created via
XComArgs.XComArgobject does not support accessing specific values of an iterableXComvalue elegantly. Did not update any occurrences in this DAG.XComArgobject does not support accessing specific values of an iterableXComvalue elegantly. Did not update any occurrences in this DAG.XComArgobject does not support accessing specific values of an iterableXComvalue elegantly. Did not update any occurrences in this DAG.XComArgobject does not support accessing specific values of an iterableXComvalue elegantly. Did not update the following occurrences:"echo {{ task_instance.xcom_pull('annotate_image')['logoAnnotations'][0]['description'] }}""echo {{ task_instance.xcom_pull('detect_text')['textAnnotations'][0] }}""echo {{ task_instance.xcom_pull('document_detect_text')['textAnnotations'][0] }}""echo {{ task_instance.xcom_pull('detect_labels')['labelAnnotations'][0] }}"All other
xcom_pull()calls have been updated.Removed explicit task dependencies that are created via
XComArgs.XComArgs.default_argspattern.XComArgobject does not support accessing specific values of an iterableXComvalue elegantly. Did not update any occurrences in this DAG.XComArgobject does not support accessing specific values of an iterableXComvalue elegantly. Did not update the following occurrence:'{{ task_instance.xcom_pull("create_sdf_download_task")["name"] }}'All other
xcom_pull()calls have been updated.Removed explicit task dependencies that are created via
XComArgs.Updated this instance to reference the correct task rather than an non-existant one:
'{{ task_instance.xcom_pull("upload_sdf_to_bigquery")}}'XComArgs.XComArgobject does not support accessing specific values of an iterableXComvalue elegantly. Did not update any occurrences in this DAG.XComArgobject does not support accessing specific values of an iterableXComvalue elegantly. Did not update the following occurrences:"{{ task_instance.xcom_pull('upload_sheet_to_gcs')[0] }}"All other
xcom_pull()calls have been updated.default_argspattern.default_argspattern.XComArgs.Refactored
grab_artifact_from_jenkins()function to take an inputXComArgsfrom a previous task.Refactored
default_argspattern.default_argspattern.default_argspattern.default_argspattern.default_argspattern.default_argspattern.default_argspattern.default_argspattern.default_argspattern.get_current_context().default_argspattern.default_argspattern.default_argspattern.default_argspattern.default_argspattern.default_argspattern.^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code change, Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in UPDATING.md.