You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "matthieucx (via GitHub)" <gi...@apache.org> on 2023/07/14 13:41:03 UTC

[GitHub] [airflow] matthieucx opened a new issue, #32603: Issue when passing the output of a traditional operator to a Task Group

matthieucx opened a new issue, #32603:
URL: https://github.com/apache/airflow/issues/32603

   ### Apache Airflow version
   
   2.6.3
   
   ### What happened
   
   Hello!
   
   When passing the output of a traditional operator to a task group, the operator is added to the task group.
   
   ![operator_tg_issue](https://github.com/apache/airflow/assets/11246353/ca99d153-e506-44f7-adf5-6eff8a6cc463)
   
   Furthermore, when expanding the task group using this output, the task group is set as upstream of the operator, which causes a cyclic dependency and fails the DAG.
   
   ![operator_tg_expansion_issue](https://github.com/apache/airflow/assets/11246353/ede4bc69-b81f-4644-903b-d7877736ddf2)
   
   As a quick fix, adding a task that simply returns its input between the task group and the operator solves the issue.
   
   ![operator_tg_expansion_fix](https://github.com/apache/airflow/assets/11246353/a7870d76-f6c1-441f-8206-4423eb1f9406)
   
   
   
   ### What you think should happen instead
   
   The operator should not be added to the task group. It should behave like a `@task()` decorated function would.
   
   ### How to reproduce
   
   The following minimal DAG is enough to reproduce the issue:
   
   ```python
   from datetime import datetime
   
   from airflow.decorators import dag, task, task_group
   from airflow.operators.dummy_operator import DummyOperator
   
   dummy_task = DummyOperator(task_id="dummy_task")
   
   
   @task()
   def print_task(data):
       print(data)
   
   
   @task()
   def passtrough(data):
       return data
   
   
   @task_group()
   def tg(data):
       print_task(data)
   
   
   @dag(
       dag_id="expansion_classic_operator",
       schedule=None,
       catchup=False,
       start_date=datetime(2021, 1, 1),
       max_active_runs=1,
   )
   def ExpansionIssueDAG():
       data_list = passtrough(dummy_task.output)
       tg.expand(data=data_list)
   
   
   dag = ExpansionIssueDAG()
   ```
   
   ### Operating System
   
   Ubuntu 22.04.2 LTS
   
   ### Versions of Apache Airflow Providers
   
   _No response_
   
   ### Deployment
   
   Docker-Compose
   
   ### Deployment details
   
   _No response_
   
   ### Anything else
   
   I would be willing to contribute, but I haven't investigated where the issue could originate from and don't have a lot of time on my hand at the moment.
   
   ### Are you willing to submit PR?
   
   - [X] Yes I am willing to submit a PR!
   
   ### Code of Conduct
   
   - [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.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.

To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [airflow] hussein-awala closed issue #32603: Issue when passing the output of a traditional operator to a Task Group

Posted by "hussein-awala (via GitHub)" <gi...@apache.org>.
hussein-awala closed issue #32603: Issue when passing the output of a traditional operator to a Task Group
URL: https://github.com/apache/airflow/issues/32603


-- 
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.

To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [airflow] jens-scheffler-bosch commented on issue #32603: Issue when passing the output of a traditional operator to a Task Group

Posted by "jens-scheffler-bosch (via GitHub)" <gi...@apache.org>.
jens-scheffler-bosch commented on issue #32603:
URL: https://github.com/apache/airflow/issues/32603#issuecomment-1636877961

   I assume this is rather a modelling issue than a problem with a task mapping logic. In your example the `dummy_task` is defined outside the DAG context and can not be correctly associated. Try it to be defined within the DAG scope, then it works, for example below:
   
   ```
   from datetime import datetime
   
   from airflow.decorators import dag, task, task_group
   from airflow.operators.dummy_operator import DummyOperator
   
   @task()
   def print_task(data):
       print(data)
   
   @task_group()
   def tg(data):
       print_task(data)
   
   @dag(
       dag_id="issue_32603",
       schedule=None,
       catchup=False,
       start_date=datetime(2021, 1, 1),
       max_active_runs=1,
   )
   def ExpansionIssueDAG():
       dummy_task = DummyOperator(task_id="dummy_task")
       tg.expand(data=dummy_task.output)
   
   
   dag = ExpansionIssueDAG()
   ```
   
   Result is:
   ![Screenshot_2023-07-15_22-13-24](https://github.com/apache/airflow/assets/95105677/7d5037ed-1aea-4daf-9de0-d31231a4c309)
   


-- 
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.

To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [airflow] hussein-awala commented on issue #32603: Issue when passing the output of a traditional operator to a Task Group

Posted by "hussein-awala (via GitHub)" <gi...@apache.org>.
hussein-awala commented on issue #32603:
URL: https://github.com/apache/airflow/issues/32603#issuecomment-1636888279

   > However, I observed another issue (it could be an expected behavior), where the tg fails when I expend it using the output of an empty task.
   
   That's because EmptyOperator (or DummyOperator) doesn't push any xcom, so we cannot use its output.
   
   I close the issue because it is not a bug, but a modelling issue.


-- 
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.

To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [airflow] matthieucx commented on issue #32603: Issue when passing the output of a traditional operator to a Task Group

Posted by "matthieucx (via GitHub)" <gi...@apache.org>.
matthieucx commented on issue #32603:
URL: https://github.com/apache/airflow/issues/32603#issuecomment-1637056115

   Thanks a lot for the correction.
   
   To anyone who ends up here having a similar issue, from what I understand:
   
   When you create a `@task()` decorated function, you create a new task template.
   When you call the function inside tha dag context, you create a task.
   
   When you call a traditional operator, you directly create a task, which is why you have to call the operator in the DAG context.
   
   Those are relevant :
   [Tasks Doc](https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/tasks.html#tasks)
   [Operators Doc](https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/operators.html#operators)
   


-- 
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.

To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [airflow] hussein-awala commented on issue #32603: Issue when passing the output of a traditional operator to a Task Group

Posted by "hussein-awala (via GitHub)" <gi...@apache.org>.
hussein-awala commented on issue #32603:
URL: https://github.com/apache/airflow/issues/32603#issuecomment-1636881730

   @jens-scheffler-bosch Indeed, I was going to write the same comment.
   
   However, I observed another issue (it could be an expected behavior), where the tg fails when I expend it using the output of an empty task.


-- 
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.

To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org