You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "keomgak (via GitHub)" <gi...@apache.org> on 2023/07/24 05:29:34 UTC
[GitHub] [airflow] keomgak opened a new issue, #32793: Airflow2 : When skipping by BranchPythonOperator with taskgroup, all downstream tasks is skipping.
keomgak opened a new issue, #32793:
URL: https://github.com/apache/airflow/issues/32793
### Apache Airflow version
Other Airflow 2 version (please specify below)
### What happened
I'm testing taskgroup in airflow2. My dag is like this,,
```
workflow_start_task
child_taskgroup
\first_task
\decision(BranchPythonOperator)
\start_job_task
\grandchild1_taskgroup
\\first_task
\\decision(BranchPythonOperator) <--------- point1
\\start_job_task
\\run_job_task
\\success_job_task
\\skip_end <---------------- point2
\grandchild2_taskgroup
\\first_task
\\decision(BranchPythonOperator)
\\start_job_task
\\run_job_task
\\success_job_task
\\skip_end
\success_job_task
\skip_end
workflow_end_task
```
When BranchPythonOperator(point 1) -> 'skip_end'(point2) case is occured, all downstream tasks are skipped.
The result is shown in the screenshot below.
I think the grandchild2_taskgroup task and the remaining tasks of the child_taskgroup should run, but I don't know why it is skipped.
### What you think should happen instead
_No response_
### How to reproduce
Here is my test code.
test_taskgroup.py
```
from datetime import datetime
from airflow import DAG
from airflow.operators.bash import BashOperator
from test_child_taskgroup import Child_Dag
with DAG(
dag_id="test_taskgroup",
schedule_interval="@daily",
start_date=datetime(2023, 7, 1),
end_date=datetime(2023, 7, 1),
default_args={"retries": 0},
) as dag:
start_task = BashOperator(
task_id='workflow_start',
bash_command='echo 1',
)
child_task = Child_Dag(dag.dag_id).get_dag()
end_task = BashOperator(
task_id='workflow_end',
bash_command='echo 1',
)
start_task >> child_task >> end_task
```
test_child_taskgroup.py
```
from airflow.models import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import BranchPythonOperator
from airflow.utils.task_group import TaskGroup
from test_grandchild_taskgroup import Grandchild_Dag
def start_decision():
return "child.start_job_task"
class Child_Dag:
def __init__(self, parent_dag_id):
self.parent_dag_id = parent_dag_id
self.workflow_id = "child"
def get_dag(self):
dag = DAG(
'%s.%s' % (self.parent_dag_id, self.workflow_id),
)
with TaskGroup(self.workflow_id) as child_dag:
first_task = BashOperator(
task_id='first_task',
bash_command='echo 1'
)
start_job_task = BashOperator(
task_id='start_job_task',
bash_command='echo 1'
)
grandchild1_id = 'grandchild1_task'
grandchild1_task = Grandchild_Dag(parent_dag_id=self.workflow_id, dataset_id=grandchild1_id).get_dag()
grandchild2_id = 'grandchild2_task'
grandchild2_task = Grandchild_Dag(parent_dag_id=self.workflow_id, dataset_id=grandchild2_id).get_dag()
success_job_task = BashOperator(
task_id='success_job_task',
bash_command='echo 1'
)
skip_end = BashOperator(
task_id='skip_end',
bash_command='echo 1',
)
decision = BranchPythonOperator(
task_id='decision',
python_callable=start_decision,
)
first_task >> decision >> start_job_task >> grandchild1_task >> grandchild2_task >> success_job_task
decision >> skip_end
return child_dag
```
test_grandchild_taskgroup.py
```
from airflow.models import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import BranchPythonOperator
from airflow.utils.task_group import TaskGroup
def start_decision(dataset_id, **kwargs):
return 'child.' + dataset_id + '.skip_end'
class Grandchild_Dag:
def __init__(self, parent_dag_id, dataset_id):
self.parent_dag_id = parent_dag_id
self.dataset_id = dataset_id
def get_dag(self):
dag = DAG(
'%s.%s' % (self.parent_dag_id, self.dataset_id),
)
with TaskGroup(self.dataset_id) as grandchild_dag:
first_task = BashOperator(
task_id='first_task',
bash_command='echo 1'
)
start_job_task = BashOperator(
task_id='start_job_task',
bash_command='echo 1'
)
run_job_task = BashOperator(
task_id='run_job_task',
bash_command='echo 1'
)
success_job_task = BashOperator(
task_id='success_job_task',
bash_command='echo 1'
)
skip_end = BashOperator(
task_id='skip_end',
bash_command='echo 1',
)
decision = BranchPythonOperator(
task_id='decision',
python_callable=start_decision,
op_kwargs={'dataset_id': self.dataset_id},
)
first_task >> decision >> start_job_task >> run_job_task >> success_job_task
decision >> skip_end
return grandchild_dag
```
### Operating System
ubuntu 22.04
### Versions of Apache Airflow Providers
composer 2.1.8 + airflow 2.4.3
composer 2.3.3 + airflow 2.5.1
### Deployment
Google Cloud Composer
### Deployment details
_No response_
### Anything else
_No response_
### Are you willing to submit PR?
- [ ] 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] yquaziii commented on issue #32793: Airflow2 : When skipping by BranchPythonOperator with taskgroup, all downstream tasks is skipping.
Posted by "yquaziii (via GitHub)" <gi...@apache.org>.
yquaziii commented on issue #32793:
URL: https://github.com/apache/airflow/issues/32793#issuecomment-1648305491
Are you using depends_on_past=True in sub_dag.? if that's not the case then the next task group should have executed.
--
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 closed issue #32793: Airflow2 : When skipping by BranchPythonOperator with taskgroup, all downstream tasks is skipping.
Posted by "hussein-awala (via GitHub)" <gi...@apache.org>.
hussein-awala closed issue #32793: Airflow2 : When skipping by BranchPythonOperator with taskgroup, all downstream tasks is skipping.
URL: https://github.com/apache/airflow/issues/32793
--
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 #32793: Airflow2 : When skipping by BranchPythonOperator with taskgroup, all downstream tasks is skipping.
Posted by "jens-scheffler-bosch (via GitHub)" <gi...@apache.org>.
jens-scheffler-bosch commented on issue #32793:
URL: https://github.com/apache/airflow/issues/32793#issuecomment-1648365888
Have you tried to model a `Dummy/EmptyOperator` in between the two task groups as a "merge point"?
--
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 #32793: Airflow2 : When skipping by BranchPythonOperator with taskgroup, all downstream tasks is skipping.
Posted by "hussein-awala (via GitHub)" <gi...@apache.org>.
hussein-awala commented on issue #32793:
URL: https://github.com/apache/airflow/issues/32793#issuecomment-1648680979
The default `trigger_rule` in Airflow is `all_success` (check the [doc](https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/dags.html#trigger-rules)), so when one of the upstream tasks finishes with a state `failed` or `upstream_failed`, the task state will passes to `upstream_failed`, and when one of them finishes with `skipped` state (without any fail in the others), the task state passes to `skipped` too.
To achieve your goal, you can set the `trigger_rule` to `none_failed` in the task `first_task` of the second task group.
I close the issue, and we can reopen it if needed.
--
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] keomgak commented on issue #32793: Airflow2 : When skipping by BranchPythonOperator with taskgroup, all downstream tasks is skipping.
Posted by "keomgak (via GitHub)" <gi...@apache.org>.
keomgak commented on issue #32793:
URL: https://github.com/apache/airflow/issues/32793#issuecomment-1648750363
@hussein-awala Thank you for the kind explanation.
I changed the trigger rule and my code works.
--
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