You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by GitBox <gi...@apache.org> on 2020/11/07 12:08:17 UTC
[GitHub] [airflow] VBhojawala opened a new pull request #12159: @taskgroup decorator implementation [https://github.com/apache/airflow/issues/11870 ]
VBhojawala opened a new pull request #12159:
URL: https://github.com/apache/airflow/pull/12159
New Feature of @taskgroup decorator implementation
closes:#11870
Added a decorator function which can be used to create TaskGroup for python function.
I have tested it with following example :
```
def generate_value():
"""Dummy function"""
return "Bring me a shrubbery!"
@task()
def print_value(value):
"""Dummy function"""
ctx = get_current_context()
log.info("Print Value: %s (at %s)", value, ctx['ts'])
return value
@task()
def log_value(value):
"""Dummy function"""
ctx = get_current_context()
log.info("Log Value: %s (at %s)", value, ctx['ts'])
@taskgroup(group_id='t3_task_1')
def t3_task_1(value):
op1 = print_value(value)
log_value(op1)
with DAG(
dag_id='T3_task_group_deco',
default_args={'owner': 'airflow'},
start_date=days_ago(2),
schedule_interval=None,
tags=['example'],
) as dag:
task1 = PythonOperator(
task_id='generate_value',
python_callable=generate_value,
)
t3_task_1(task1.output)
```
<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
In case of existing issue, reference it using one of the following:
closes: #ISSUE
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst#pull-request-guidelines)** for more information.
In case of fundamental code change, Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvements+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in [UPDATING.md](https://github.com/apache/airflow/blob/master/UPDATING.md).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] boring-cyborg[bot] commented on pull request #12159: @taskgroup decorator implementation [https://github.com/apache/airflow/issues/11870 ]
Posted by GitBox <gi...@apache.org>.
boring-cyborg[bot] commented on pull request #12159:
URL: https://github.com/apache/airflow/pull/12159#issuecomment-723438144
Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contribution Guide (https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst)
Here are some useful points:
- Pay attention to the quality of your code (flake8, pylint and type annotations). Our [pre-commits]( https://github.com/apache/airflow/blob/master/STATIC_CODE_CHECKS.rst#prerequisites-for-pre-commit-hooks) will help you with that.
- In case of a new feature add useful documentation (in docstrings or in `docs/` directory). Adding a new operator? Check this short [guide](https://github.com/apache/airflow/blob/master/docs/howto/custom-operator.rst) Consider adding an example DAG that shows how users should use it.
- Consider using [Breeze environment](https://github.com/apache/airflow/blob/master/BREEZE.rst) for testing locally, itโs a heavy docker but it ships with a working Airflow and a lot of integrations.
- Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
- Please follow [ASF Code of Conduct](https://www.apache.org/foundation/policies/conduct) for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
- Be sure to read the [Airflow Coding style]( https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst#coding-style-and-best-practices).
Apache Airflow is a community-driven project and together we are making it better ๐.
In case of doubts contact the developers at:
Mailing List: dev@airflow.apache.org
Slack: https://s.apache.org/airflow-slack
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] VBhojawala edited a comment on pull request #12159: @taskgroup decorator implementation [https://github.com/apache/airflow/issues/11870 ]
Posted by GitBox <gi...@apache.org>.
VBhojawala edited a comment on pull request #12159:
URL: https://github.com/apache/airflow/pull/12159#issuecomment-723786153
> Thanks for proposing this. I think it works for small examples. However I have some questions regarding how this should work for larger examples. For instance, if we wanna put two or more tasks in the same `TaskGroup`, or a `TaskGroup` nested in another `TaskGroup`, can that be done with with `@task_group`?
>
> The reason I'm asking is that the contextmanager syntax is quite natural for creating nested structures (which is what `TaskGroup` is). The decorator syntax can probably be tweaked to do something close enough, but it's not as straightforward. If you have a good way of doing that, maybe demonstrate that in the example?
Yes it does works because we have used the same context manager of TaskGroup, i am writing test cases for that also. Here is the example
``` python
from airflow.models.dag import DAG
from airflow.operators.python import task
from airflow.utils.dates import days_ago
from airflow.utils.task_group import taskgroup
@task()
def task_start():
"""Dummy Task which is First Task of Dag """
return '[Task_start]'
@task()
def task_end():
"""Dummy Task which is Last Task of Dag"""
print(f'[ Task_End ]')
@task
def task_1(value):
""" Dummy Task1"""
return f'[ Task1 {value} ]'
@task
def task_2(value):
""" Dummy Task2"""
print(f'[ Task2 {value} ]')
@task
def task_3(value):
""" Dummy Task3"""
return f'[ Task3 {value} ]'
@task
def task_4(value):
""" Dummy Task3"""
print(f'[ Task4 {value} ]')
# Creating TaskGroups
@taskgroup(group_id='section_1')
def section_1(value):
""" TaskGroup for grouping related Tasks"""
@taskgroup(group_id='section_2')
def section_2(value2):
""" Nested TaskGroup for grouping related Tasks"""
return task_4(task_3(value2))
op1 = task_2(task_1(value))
return section_2(op1)
with DAG(dag_id="example_nested_task_group_decorator", start_date=days_ago(2), tags=["example"]) as dag:
t1 = task_start()
s1 = section_1(t1)
s1.set_downstream(task_end())
```
@turbaszek Should we include above example in example_dags folder
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] VBhojawala commented on pull request #12159: @taskgroup decorator implementation [https://github.com/apache/airflow/issues/11870 ]
Posted by GitBox <gi...@apache.org>.
VBhojawala commented on pull request #12159:
URL: https://github.com/apache/airflow/pull/12159#issuecomment-724436894
Re-raising pull request.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] VBhojawala edited a comment on pull request #12159: @taskgroup decorator implementation [https://github.com/apache/airflow/issues/11870 ]
Posted by GitBox <gi...@apache.org>.
VBhojawala edited a comment on pull request #12159:
URL: https://github.com/apache/airflow/pull/12159#issuecomment-723786153
> Thanks for proposing this. I think it works for small examples. However I have some questions regarding how this should work for larger examples. For instance, if we wanna put two or more tasks in the same `TaskGroup`, or a `TaskGroup` nested in another `TaskGroup`, can that be done with with `@task_group`?
>
> The reason I'm asking is that the contextmanager syntax is quite natural for creating nested structures (which is what `TaskGroup` is). The decorator syntax can probably be tweaked to do something close enough, but it's not as straightforward. If you have a good way of doing that, maybe demonstrate that in the example?
Yes it does works because we have used the same context manager of TaskGroup, i am writing test cases for that also. Here is the example
``` python
from airflow.models.dag import DAG
from airflow.operators.python import task
from airflow.utils.dates import days_ago
from airflow.utils.task_group import taskgroup
@task()
def task_start():
"""Dummy Task which is First Task of Dag """
return '[Task_start]'
@task()
def task_end():
"""Dummy Task which is Last Task of Dag"""
print(f'[ Task_End ]')
@task
def task_1(value):
""" Dummy Task1"""
return f'[ Task1 {value} ]'
@task
def task_2(value):
""" Dummy Task2"""
print(f'[ Task2 {value} ]')
@task
def task_3(value):
""" Dummy Task3"""
return f'[ Task3 {value} ]'
@task
def task_4(value):
""" Dummy Task3"""
print(f'[ Task4 {value} ]')
# Creating TaskGroups
@taskgroup(group_id='section_1')
def section_1(value):
""" TaskGroup for grouping related Tasks"""
@taskgroup(group_id='section_2')
def section_2(value2):
""" Nested TaskGroup for grouping related Tasks"""
return task_4(task_3(value2))
op1 = task_2(task_1(value))
return section_2(op1)
with DAG(dag_id="example_nested_task_group_decorator", start_date=days_ago(2), tags=["example"]) as dag:
t1 = task_start()
s1 = section_1(t1)
s1.set_downstream(task_end())
```
@turbaszek Should we include above example in example_dags folder?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] yuqian90 commented on pull request #12159: @taskgroup decorator implementation [https://github.com/apache/airflow/issues/11870 ]
Posted by GitBox <gi...@apache.org>.
yuqian90 commented on pull request #12159:
URL: https://github.com/apache/airflow/pull/12159#issuecomment-723760131
Thanks for proposing this. I think it works for small examples. However I have some questions regarding how this should work for larger examples. For instance, if we wanna put two or more tasks in the same `TaskGroup`, can that be done with with `@task_group`?
The reason I'm asking is that the contextmanager syntax is quite natural for creating nested structures (which is what `TaskGroup` is). The decorator syntax can probably be tweaked to do something close enough, but it's not as straightforward. If you have a good way of doing that, maybe demonstrate that in the example?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] VBhojawala edited a comment on pull request #12159: @taskgroup decorator implementation [https://github.com/apache/airflow/issues/11870 ]
Posted by GitBox <gi...@apache.org>.
VBhojawala edited a comment on pull request #12159:
URL: https://github.com/apache/airflow/pull/12159#issuecomment-723786153
> Thanks for proposing this. I think it works for small examples. However I have some questions regarding how this should work for larger examples. For instance, if we wanna put two or more tasks in the same `TaskGroup`, or a `TaskGroup` nested in another `TaskGroup`, can that be done with with `@task_group`?
>
> The reason I'm asking is that the contextmanager syntax is quite natural for creating nested structures (which is what `TaskGroup` is). The decorator syntax can probably be tweaked to do something close enough, but it's not as straightforward. If you have a good way of doing that, maybe demonstrate that in the example?
Yes it does works because we have used the same context manager of TaskGroup, i am writing test cases for that also. Here is the example
``` python
from airflow.models.dag import DAG
from airflow.operators.python import task
from airflow.utils.dates import days_ago
from airflow.utils.task_group import taskgroup
@task()
def task_start():
"""Dummy Task which is First Task of Dag """
return '[Task_start]'
@task()
def task_end():
"""Dummy Task which is Last Task of Dag"""
print(f'[ Task_End ]')
@task
def task_1(value):
""" Dummy Task1"""
return f'[ Task1 {value} ]'
@task
def task_2(value):
""" Dummy Task2"""
print(f'[ Task2 {value} ]')
@task
def task_3(value):
""" Dummy Task3"""
return f'[ Task3 {value} ]'
@task
def task_4(value):
""" Dummy Task3"""
print(f'[ Task4 {value} ]')
# Creating TaskGroups
@taskgroup(group_id='section_1')
def section_1(value):
""" TaskGroup for grouping related Tasks"""
@taskgroup(group_id='section_2')
def section_2(value2):
""" Nested TaskGroup for grouping related Tasks"""
return task_4(task_3(value2))
op1 = task_2(task_1(value))
return section_2(op1)
def test_build_task_group_context_manager():
with DAG(dag_id="example_nested_task_group_decorator", start_date=days_ago(2), tags=["example"]) as dag:
t1 = task_start()
s1 = section_1(t1)
s1.set_downstream(task_end())
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] yuqian90 edited a comment on pull request #12159: @taskgroup decorator implementation [https://github.com/apache/airflow/issues/11870 ]
Posted by GitBox <gi...@apache.org>.
yuqian90 edited a comment on pull request #12159:
URL: https://github.com/apache/airflow/pull/12159#issuecomment-723760131
Thanks for proposing this. I think it works for small examples. However I have some questions regarding how this should work for larger examples. For instance, if we wanna put two or more tasks in the same `TaskGroup`, or a `TaskGroup` nested in another `TaskGroup`, can that be done with with `@task_group`?
The reason I'm asking is that the contextmanager syntax is quite natural for creating nested structures (which is what `TaskGroup` is). The decorator syntax can probably be tweaked to do something close enough, but it's not as straightforward. If you have a good way of doing that, maybe demonstrate that in the example?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] VBhojawala commented on pull request #12159: @taskgroup decorator implementation [https://github.com/apache/airflow/issues/11870 ]
Posted by GitBox <gi...@apache.org>.
VBhojawala commented on pull request #12159:
URL: https://github.com/apache/airflow/pull/12159#issuecomment-723802470
> > Thanks for proposing this. I think it works for small examples. However I have some questions regarding how this should work for larger examples. For instance, if we wanna put two or more tasks in the same `TaskGroup`, or a `TaskGroup` nested in another `TaskGroup`, can that be done with with `@task_group`?
> > The reason I'm asking is that the contextmanager syntax is quite natural for creating nested structures (which is what `TaskGroup` is). The decorator syntax can probably be tweaked to do something close enough, but it's not as straightforward. If you have a good way of doing that, maybe demonstrate that in the example?
>
> Yes it does works because we have used the same context manager of TaskGroup, i am writing test cases for that also. Here is the example
>
> ```python
> from airflow.models.dag import DAG
> from airflow.operators.python import task
> from airflow.utils.dates import days_ago
> from airflow.utils.task_group import taskgroup
>
>
> @task()
> def task_start():
> """Dummy Task which is First Task of Dag """
> return '[Task_start]'
>
>
> @task()
> def task_end():
> """Dummy Task which is Last Task of Dag"""
> print(f'[ Task_End ]')
>
>
> @task
> def task_1(value):
> """ Dummy Task1"""
> return f'[ Task1 {value} ]'
>
>
> @task
> def task_2(value):
> """ Dummy Task2"""
> print(f'[ Task2 {value} ]')
>
>
> @task
> def task_3(value):
> """ Dummy Task3"""
> return f'[ Task3 {value} ]'
>
>
> @task
> def task_4(value):
> """ Dummy Task3"""
> print(f'[ Task4 {value} ]')
>
>
> # Creating TaskGroups
> @taskgroup(group_id='section_1')
> def section_1(value):
> """ TaskGroup for grouping related Tasks"""
>
> @taskgroup(group_id='section_2')
> def section_2(value2):
> """ Nested TaskGroup for grouping related Tasks"""
> return task_4(task_3(value2))
>
> op1 = task_2(task_1(value))
> return section_2(op1)
>
>
> with DAG(dag_id="example_nested_task_group_decorator", start_date=days_ago(2), tags=["example"]) as dag:
>
> t1 = task_start()
> s1 = section_1(t1)
> s1.set_downstream(task_end())
> ```
@turbaszek Should we include above example in example_dags folder?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #12159: @taskgroup decorator implementation [https://github.com/apache/airflow/issues/11870 ]
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #12159:
URL: https://github.com/apache/airflow/pull/12159#issuecomment-723874069
[The Workflow run](https://github.com/apache/airflow/actions/runs/353564402) is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] VBhojawala commented on pull request #12159: @taskgroup decorator implementation [https://github.com/apache/airflow/issues/11870 ]
Posted by GitBox <gi...@apache.org>.
VBhojawala commented on pull request #12159:
URL: https://github.com/apache/airflow/pull/12159#issuecomment-723860462
Hello @turbaszek,
I have added test cases for taskgroup decorator. Kindly review it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] VBhojawala edited a comment on pull request #12159: @taskgroup decorator implementation [https://github.com/apache/airflow/issues/11870 ]
Posted by GitBox <gi...@apache.org>.
VBhojawala edited a comment on pull request #12159:
URL: https://github.com/apache/airflow/pull/12159#issuecomment-723786153
> Thanks for proposing this. I think it works for small examples. However I have some questions regarding how this should work for larger examples. For instance, if we wanna put two or more tasks in the same `TaskGroup`, or a `TaskGroup` nested in another `TaskGroup`, can that be done with with `@task_group`?
>
> The reason I'm asking is that the contextmanager syntax is quite natural for creating nested structures (which is what `TaskGroup` is). The decorator syntax can probably be tweaked to do something close enough, but it's not as straightforward. If you have a good way of doing that, maybe demonstrate that in the example?
Yes it does works because we have used the same context manager of TaskGroup, i am writing test cases for that also. Here is the example
``` python
from airflow.models.dag import DAG
from airflow.operators.python import task
from airflow.utils.dates import days_ago
from airflow.utils.task_group import taskgroup
from airflow.www.views import task_group_to_dict
# Creating Tasks
@task()
def task_start():
"""Dummy Task which is First Task of Dag """
return '[Task_start]'
@task()
def task_end1():
"""Dummy Task which is Last Task of Dag"""
print(f'[ Task_End 1 ]')
@task()
def task_end2():
"""Dummy Task which is Last Task of Dag"""
print(f'[ Task_End 2 ]')
@task
def task_1(value):
""" Dummy Task1"""
return f'[ Task1 {value} ]'
@task
def task_2(value):
""" Dummy Task2"""
print(f'[ Task2 {value} ]')
@task
def task_3(value):
""" Dummy Task3"""
return f'[ Task3 {value} ]'
@task
def task_4(value):
""" Dummy Task3"""
print(f'[ Task4 {value} ]')
# Creating TaskGroups
@taskgroup(group_id='section_1')
def section_1(value):
""" TaskGroup for grouping related Tasks"""
@taskgroup(group_id='section_2')
def section_2(value2):
"""Nested TaskGroup for grouping related Tasks"""
return task_4(task_3(value2))
op1 = task_2(task_1(value))
return section_2(op1)
# Executing Tasks and TaskGroups
with DAG(dag_id="example_nested_task_group_decorator", start_date=days_ago(2), tags=["example"]) as dag:
t1 = task_start()
s1 = section_1(t1)
s1.set_downstream(task_end1())
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek commented on a change in pull request #12159: @taskgroup decorator implementation [https://github.com/apache/airflow/issues/11870 ]
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #12159:
URL: https://github.com/apache/airflow/pull/12159#discussion_r519229393
##########
File path: docs/concepts.rst
##########
@@ -1017,6 +1017,32 @@ This animated gif shows the UI interactions. TaskGroups are expanded or collapse
.. image:: img/task_group.gif
+@taskgroup decorator
+--------------------
+
+TaskGroup can be created using @taskgroup decorator, it takes one mandatory argument 'group_id' which is same as constructor of TaskGroup class. It works exactly same as creating TaskGroup using context manager 'with TaskGroup('groupid') as section:'.
Review comment:
```suggestion
TaskGroup can be created using ``@taskgroup decorator``, it takes one mandatory argument ``group_id`` which is same as constructor of TaskGroup class. It works exactly same as creating TaskGroup using context manager ``with TaskGroup('groupid') as section:``.
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #12159: @taskgroup decorator implementation [https://github.com/apache/airflow/issues/11870 ]
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #12159:
URL: https://github.com/apache/airflow/pull/12159#issuecomment-723471297
[The Workflow run](https://github.com/apache/airflow/actions/runs/351328035) is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #12159: @taskgroup decorator implementation [https://github.com/apache/airflow/issues/11870 ]
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #12159:
URL: https://github.com/apache/airflow/pull/12159#issuecomment-723438846
[The Workflow run](https://github.com/apache/airflow/actions/runs/351041990) is cancelling this PR. Building images for the PR has failed. Follow the the workflow link to check the reason.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] VBhojawala commented on pull request #12159: @taskgroup decorator implementation [https://github.com/apache/airflow/issues/11870 ]
Posted by GitBox <gi...@apache.org>.
VBhojawala commented on pull request #12159:
URL: https://github.com/apache/airflow/pull/12159#issuecomment-723786153
> Thanks for proposing this. I think it works for small examples. However I have some questions regarding how this should work for larger examples. For instance, if we wanna put two or more tasks in the same `TaskGroup`, or a `TaskGroup` nested in another `TaskGroup`, can that be done with with `@task_group`?
>
> The reason I'm asking is that the contextmanager syntax is quite natural for creating nested structures (which is what `TaskGroup` is). The decorator syntax can probably be tweaked to do something close enough, but it's not as straightforward. If you have a good way of doing that, maybe demonstrate that in the example?
Yes it does works because we have used the same context manager used TaskGroup, i am writing test cases for that also. Here is the example
``` python
from airflow.models.dag import DAG
from airflow.operators.python import task
from airflow.utils.dates import days_ago
from airflow.utils.task_group import taskgroup
# Creating Tasks
from airflow.www.views import task_group_to_dict
@task()
def task_start():
"""Dummy Task which is First Task of Dag """
return '[Task_start]'
@task()
def task_end1():
"""Dummy Task which is Last Task of Dag"""
print(f'[ Task_End 1 ]')
@task()
def task_end2():
"""Dummy Task which is Last Task of Dag"""
print(f'[ Task_End 2 ]')
@task
def task_1(value):
""" Dummy Task1"""
return f'[ Task1 {value} ]'
@task
def task_2(value):
""" Dummy Task2"""
print(f'[ Task2 {value} ]')
@task
def task_3(value):
""" Dummy Task3"""
return f'[ Task3 {value} ]'
@task
def task_4(value):
""" Dummy Task3"""
print(f'[ Task4 {value} ]')
# Creating TaskGroups
@taskgroup(group_id='section_1')
def section_1(value):
""" TaskGroup for grouping related Tasks"""
@taskgroup(group_id='section_2')
def section_2(value2):
"""Nested TaskGroup for grouping related Tasks"""
return task_4(task_3(value2))
op1 = task_2(task_1(value))
return section_2(op1)
# Executing Tasks and TaskGroups
with DAG(dag_id="example_nested_task_group_decorator", start_date=days_ago(2), tags=["example"]) as dag:
t1 = task_start()
s1 = section_1(t1)
s1.set_downstream(task_end1())
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] VBhojawala edited a comment on pull request #12159: @taskgroup decorator implementation [https://github.com/apache/airflow/issues/11870 ]
Posted by GitBox <gi...@apache.org>.
VBhojawala edited a comment on pull request #12159:
URL: https://github.com/apache/airflow/pull/12159#issuecomment-723786153
> Thanks for proposing this. I think it works for small examples. However I have some questions regarding how this should work for larger examples. For instance, if we wanna put two or more tasks in the same `TaskGroup`, or a `TaskGroup` nested in another `TaskGroup`, can that be done with with `@task_group`?
>
> The reason I'm asking is that the contextmanager syntax is quite natural for creating nested structures (which is what `TaskGroup` is). The decorator syntax can probably be tweaked to do something close enough, but it's not as straightforward. If you have a good way of doing that, maybe demonstrate that in the example?
Yes it does works because we have used the same context manager of TaskGroup, i am writing test cases for that also. Here is the example
``` python
from airflow.models.dag import DAG
from airflow.operators.python import task
from airflow.utils.dates import days_ago
from airflow.utils.task_group import taskgroup
@task()
def task_start():
"""Dummy Task which is First Task of Dag """
return '[Task_start]'
@task()
def task_end():
"""Dummy Task which is Last Task of Dag"""
print(f'[ Task_End ]')
@task
def task_1(value):
""" Dummy Task1"""
return f'[ Task1 {value} ]'
@task
def task_2(value):
""" Dummy Task2"""
print(f'[ Task2 {value} ]')
@task
def task_3(value):
""" Dummy Task3"""
return f'[ Task3 {value} ]'
@task
def task_4(value):
""" Dummy Task4"""
print(f'[ Task4 {value} ]')
# Creating TaskGroups
@taskgroup(group_id='section_1')
def section_1(value):
""" TaskGroup for grouping related Tasks"""
@taskgroup(group_id='section_2')
def section_2(value2):
""" Nested TaskGroup for grouping related Tasks"""
return task_4(task_3(value2))
op1 = task_2(task_1(value))
return section_2(op1)
with DAG(dag_id="example_nested_task_group_decorator", start_date=days_ago(2), tags=["example"]) as dag:
t1 = task_start()
s1 = section_1(t1)
s1.set_downstream(task_end())
```
@turbaszek Should we include above example in example_dags folder?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek edited a comment on pull request #12159: @taskgroup decorator implementation [https://github.com/apache/airflow/issues/11870 ]
Posted by GitBox <gi...@apache.org>.
turbaszek edited a comment on pull request #12159:
URL: https://github.com/apache/airflow/pull/12159#issuecomment-723440639
CC @yuqian90 @casassg
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] VBhojawala edited a comment on pull request #12159: @taskgroup decorator implementation [https://github.com/apache/airflow/issues/11870 ]
Posted by GitBox <gi...@apache.org>.
VBhojawala edited a comment on pull request #12159:
URL: https://github.com/apache/airflow/pull/12159#issuecomment-723786153
> Thanks for proposing this. I think it works for small examples. However I have some questions regarding how this should work for larger examples. For instance, if we wanna put two or more tasks in the same `TaskGroup`, or a `TaskGroup` nested in another `TaskGroup`, can that be done with with `@task_group`?
>
> The reason I'm asking is that the contextmanager syntax is quite natural for creating nested structures (which is what `TaskGroup` is). The decorator syntax can probably be tweaked to do something close enough, but it's not as straightforward. If you have a good way of doing that, maybe demonstrate that in the example?
Yes it does works because we have used the same context manager of TaskGroup, i am writing test cases for that also. Here is the example
``` python
from airflow.models.dag import DAG
from airflow.operators.python import task
from airflow.utils.dates import days_ago
from airflow.utils.task_group import taskgroup
@task()
def task_start():
"""Dummy Task which is First Task of Dag """
return '[Task_start]'
@task()
def task_end():
"""Dummy Task which is Last Task of Dag"""
print(f'[ Task_End ]')
@task
def task_1(value):
""" Dummy Task1"""
return f'[ Task1 {value} ]'
@task
def task_2(value):
""" Dummy Task2"""
print(f'[ Task2 {value} ]')
@task
def task_3(value):
""" Dummy Task3"""
return f'[ Task3 {value} ]'
@task
def task_4(value):
""" Dummy Task3"""
print(f'[ Task4 {value} ]')
# Creating TaskGroups
@taskgroup(group_id='section_1')
def section_1(value):
""" TaskGroup for grouping related Tasks"""
@taskgroup(group_id='section_2')
def section_2(value2):
""" TaskGroup for grouping related Tasks"""
return task_4(task_3(value2))
op1 = task_2(task_1(value))
return section_2(op1)
def test_build_task_group_context_manager():
with DAG(dag_id="example_nested_task_group_decorator", start_date=days_ago(2), tags=["example"]) as dag:
t1 = task_start()
s1 = section_1(t1)
s1.set_downstream(task_end())
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek commented on pull request #12159: @taskgroup decorator implementation [https://github.com/apache/airflow/issues/11870 ]
Posted by GitBox <gi...@apache.org>.
turbaszek commented on pull request #12159:
URL: https://github.com/apache/airflow/pull/12159#issuecomment-723440639
CC @yuqian90
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] VBhojawala commented on pull request #12159: @taskgroup decorator implementation [https://github.com/apache/airflow/issues/11870 ]
Posted by GitBox <gi...@apache.org>.
VBhojawala commented on pull request #12159:
URL: https://github.com/apache/airflow/pull/12159#issuecomment-723899274
To simplify the `@taskgroup ` decorator syntax, made group_id parameter optional . If taskgroup is only created with @taskgroup() decorator it will automatically assign group_id as function name.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #12159: @taskgroup decorator implementation [https://github.com/apache/airflow/issues/11870 ]
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #12159:
URL: https://github.com/apache/airflow/pull/12159#issuecomment-723972861
[The Workflow run](https://github.com/apache/airflow/actions/runs/353843754) is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek commented on a change in pull request #12159: @taskgroup decorator implementation [https://github.com/apache/airflow/issues/11870 ]
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #12159:
URL: https://github.com/apache/airflow/pull/12159#discussion_r519229443
##########
File path: docs/concepts.rst
##########
@@ -1017,6 +1017,32 @@ This animated gif shows the UI interactions. TaskGroups are expanded or collapse
.. image:: img/task_group.gif
+@taskgroup decorator
+--------------------
+
+TaskGroup can be created using @taskgroup decorator, it takes one mandatory argument 'group_id' which is same as constructor of TaskGroup class. It works exactly same as creating TaskGroup using context manager 'with TaskGroup('groupid') as section:'.
+
+.. code-block:: python
+
+ @task
+ def task_1(value):
+ return f'[ Task1 {value} ]'
+
+
+ @task
+ def task_2(value):
+ print( f'[ Task2 {value} ]')
Review comment:
```suggestion
print(f'[ Task2 {value} ]')
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek commented on a change in pull request #12159: @taskgroup decorator implementation [https://github.com/apache/airflow/issues/11870 ]
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #12159:
URL: https://github.com/apache/airflow/pull/12159#discussion_r519229524
##########
File path: docs/concepts.rst
##########
@@ -1017,6 +1017,32 @@ This animated gif shows the UI interactions. TaskGroups are expanded or collapse
.. image:: img/task_group.gif
+@taskgroup decorator
+--------------------
+
+TaskGroup can be created using @taskgroup decorator, it takes one mandatory argument 'group_id' which is same as constructor of TaskGroup class. It works exactly same as creating TaskGroup using context manager 'with TaskGroup('groupid') as section:'.
+
+.. code-block:: python
+
+ @task
+ def task_1(value):
+ return f'[ Task1 {value} ]'
+
+
+ @task
+ def task_2(value):
+ print( f'[ Task2 {value} ]')
+
+
+ @taskgroup(group_id='section_1')
+ def section_1(value):
+ return task_2(task_1(value))
+
Review comment:
I'm not sure if we need two examples. If we decide to keep both the it would be nice to add some text between them.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] VBhojawala edited a comment on pull request #12159: @taskgroup decorator implementation [https://github.com/apache/airflow/issues/11870 ]
Posted by GitBox <gi...@apache.org>.
VBhojawala edited a comment on pull request #12159:
URL: https://github.com/apache/airflow/pull/12159#issuecomment-723786153
> Thanks for proposing this. I think it works for small examples. However I have some questions regarding how this should work for larger examples. For instance, if we wanna put two or more tasks in the same `TaskGroup`, or a `TaskGroup` nested in another `TaskGroup`, can that be done with with `@task_group`?
>
> The reason I'm asking is that the contextmanager syntax is quite natural for creating nested structures (which is what `TaskGroup` is). The decorator syntax can probably be tweaked to do something close enough, but it's not as straightforward. If you have a good way of doing that, maybe demonstrate that in the example?
Yes it does works because we have used the same context manager of TaskGroup, i am writing test cases for that also. Here is the example
``` python
from airflow.models.dag import DAG
from airflow.operators.python import task
from airflow.utils.dates import days_ago
from airflow.utils.task_group import taskgroup
# Creating Tasks
from airflow.www.views import task_group_to_dict
@task()
def task_start():
"""Dummy Task which is First Task of Dag """
return '[Task_start]'
@task()
def task_end1():
"""Dummy Task which is Last Task of Dag"""
print(f'[ Task_End 1 ]')
@task()
def task_end2():
"""Dummy Task which is Last Task of Dag"""
print(f'[ Task_End 2 ]')
@task
def task_1(value):
""" Dummy Task1"""
return f'[ Task1 {value} ]'
@task
def task_2(value):
""" Dummy Task2"""
print(f'[ Task2 {value} ]')
@task
def task_3(value):
""" Dummy Task3"""
return f'[ Task3 {value} ]'
@task
def task_4(value):
""" Dummy Task3"""
print(f'[ Task4 {value} ]')
# Creating TaskGroups
@taskgroup(group_id='section_1')
def section_1(value):
""" TaskGroup for grouping related Tasks"""
@taskgroup(group_id='section_2')
def section_2(value2):
"""Nested TaskGroup for grouping related Tasks"""
return task_4(task_3(value2))
op1 = task_2(task_1(value))
return section_2(op1)
# Executing Tasks and TaskGroups
with DAG(dag_id="example_nested_task_group_decorator", start_date=days_ago(2), tags=["example"]) as dag:
t1 = task_start()
s1 = section_1(t1)
s1.set_downstream(task_end1())
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #12159: @taskgroup decorator implementation [https://github.com/apache/airflow/issues/11870 ]
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #12159:
URL: https://github.com/apache/airflow/pull/12159#issuecomment-724028787
[The Workflow run](https://github.com/apache/airflow/actions/runs/353991223) is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] VBhojawala edited a comment on pull request #12159: @taskgroup decorator implementation [https://github.com/apache/airflow/issues/11870 ]
Posted by GitBox <gi...@apache.org>.
VBhojawala edited a comment on pull request #12159:
URL: https://github.com/apache/airflow/pull/12159#issuecomment-723786153
> Thanks for proposing this. I think it works for small examples. However I have some questions regarding how this should work for larger examples. For instance, if we wanna put two or more tasks in the same `TaskGroup`, or a `TaskGroup` nested in another `TaskGroup`, can that be done with with `@task_group`?
>
> The reason I'm asking is that the contextmanager syntax is quite natural for creating nested structures (which is what `TaskGroup` is). The decorator syntax can probably be tweaked to do something close enough, but it's not as straightforward. If you have a good way of doing that, maybe demonstrate that in the example?
Yes it does works because we have used the same context manager of TaskGroup, i am writing test cases for that also. Here is the example
``` python
from airflow.models.dag import DAG
from airflow.operators.python import task
from airflow.utils.dates import days_ago
from airflow.utils.task_group import taskgroup
@task()
def task_start():
"""Dummy Task which is First Task of Dag """
return '[Task_start]'
@task()
def task_end():
"""Dummy Task which is Last Task of Dag"""
print(f'[ Task_End ]')
@task
def task_1(value):
""" Dummy Task1"""
return f'[ Task1 {value} ]'
@task
def task_2(value):
""" Dummy Task2"""
print(f'[ Task2 {value} ]')
@task
def task_3(value):
""" Dummy Task3"""
return f'[ Task3 {value} ]'
@task
def task_4(value):
""" Dummy Task3"""
print(f'[ Task4 {value} ]')
# Creating TaskGroups
@taskgroup(group_id='section_1')
def section_1(value):
""" TaskGroup for grouping related Tasks"""
@taskgroup(group_id='section_2')
def section_2(value2):
""" Nested TaskGroup for grouping related Tasks"""
return task_4(task_3(value2))
op1 = task_2(task_1(value))
return section_2(op1)
with DAG(dag_id="example_nested_task_group_decorator", start_date=days_ago(2), tags=["example"]) as dag:
t1 = task_start()
s1 = section_1(t1)
s1.set_downstream(task_end())
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] VBhojawala closed pull request #12159: @taskgroup decorator implementation [https://github.com/apache/airflow/issues/11870 ]
Posted by GitBox <gi...@apache.org>.
VBhojawala closed pull request #12159:
URL: https://github.com/apache/airflow/pull/12159
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek commented on a change in pull request #12159: @taskgroup decorator implementation [https://github.com/apache/airflow/issues/11870 ]
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #12159:
URL: https://github.com/apache/airflow/pull/12159#discussion_r519229349
##########
File path: docs/concepts.rst
##########
@@ -1017,6 +1017,32 @@ This animated gif shows the UI interactions. TaskGroups are expanded or collapse
.. image:: img/task_group.gif
+@taskgroup decorator
Review comment:
```suggestion
TaskGroup decorator
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] VBhojawala removed a comment on pull request #12159: @taskgroup decorator implementation [https://github.com/apache/airflow/issues/11870 ]
Posted by GitBox <gi...@apache.org>.
VBhojawala removed a comment on pull request #12159:
URL: https://github.com/apache/airflow/pull/12159#issuecomment-723802470
> > Thanks for proposing this. I think it works for small examples. However I have some questions regarding how this should work for larger examples. For instance, if we wanna put two or more tasks in the same `TaskGroup`, or a `TaskGroup` nested in another `TaskGroup`, can that be done with with `@task_group`?
> > The reason I'm asking is that the contextmanager syntax is quite natural for creating nested structures (which is what `TaskGroup` is). The decorator syntax can probably be tweaked to do something close enough, but it's not as straightforward. If you have a good way of doing that, maybe demonstrate that in the example?
>
> Yes it does works because we have used the same context manager of TaskGroup, i am writing test cases for that also. Here is the example
>
> ```python
> from airflow.models.dag import DAG
> from airflow.operators.python import task
> from airflow.utils.dates import days_ago
> from airflow.utils.task_group import taskgroup
>
>
> @task()
> def task_start():
> """Dummy Task which is First Task of Dag """
> return '[Task_start]'
>
>
> @task()
> def task_end():
> """Dummy Task which is Last Task of Dag"""
> print(f'[ Task_End ]')
>
>
> @task
> def task_1(value):
> """ Dummy Task1"""
> return f'[ Task1 {value} ]'
>
>
> @task
> def task_2(value):
> """ Dummy Task2"""
> print(f'[ Task2 {value} ]')
>
>
> @task
> def task_3(value):
> """ Dummy Task3"""
> return f'[ Task3 {value} ]'
>
>
> @task
> def task_4(value):
> """ Dummy Task3"""
> print(f'[ Task4 {value} ]')
>
>
> # Creating TaskGroups
> @taskgroup(group_id='section_1')
> def section_1(value):
> """ TaskGroup for grouping related Tasks"""
>
> @taskgroup(group_id='section_2')
> def section_2(value2):
> """ Nested TaskGroup for grouping related Tasks"""
> return task_4(task_3(value2))
>
> op1 = task_2(task_1(value))
> return section_2(op1)
>
>
> with DAG(dag_id="example_nested_task_group_decorator", start_date=days_ago(2), tags=["example"]) as dag:
>
> t1 = task_start()
> s1 = section_1(t1)
> s1.set_downstream(task_end())
> ```
@turbaszek Should we include above example in example_dags folder?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org