You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "Yuvraj-Dhepe (via GitHub)" <gi...@apache.org> on 2025/04/21 14:26:08 UTC
[D] DAG Tasks Creation [airflow]
GitHub user Yuvraj-Dhepe created a discussion: DAG Tasks Creation
```python
from airflow.utils.task_group import TaskGroup
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.empty import EmptyOperator
from datetime import datetime, timedelta
import sys
def foo():
func_name = sys._getframe().f_code.co_name
python_task = PythonOperator(
task_id=func_name, python_callable=run_task, op_args=[func_name]
)
return python_task
def fun():
func_name = sys._getframe().f_code.co_name
python_task = PythonOperator(
task_id=func_name, python_callable=run_task, op_args=[func_name]
)
return python_task
class Foo:
def class_foo(self, dag):
func_name = sys._getframe().f_code.co_name
with TaskGroup(f"{func_name}", dag=dag) as task:
python_task = PythonOperator(
task_id=func_name, python_callable=run_task, op_args=[func_name]
)
# python_task
return task
class Fun:
def class_fun(self, dag):
func_name = sys._getframe().f_code.co_name
python_task = PythonOperator(
task_id=func_name, python_callable=run_task, op_args=[func_name]
)
# python_task
return python_task
def run_task(x):
print(f"Hello {x}")
default_args = {
"owner": "airflow",
"depends_on_past": False,
"start_date": datetime.now(),
"email_on_failure": False,
"email_on_retry": False,
"retries": 0,
}
# NOTE: Runs the dag, but even if t2 is not in the dag dependency still it shows up in the dashboard, I couldn't understand the reason for it.
# with DAG(
# dag_id="dag_sample",
# default_args=default_args,
# description="A simple Airflow DAG",
# schedule=timedelta(days=1), # Explicit schedule
# catchup=False,
# tags=["D1"],
# ) as dag:
# start = EmptyOperator(task_id="start")
# t1 = foo()
# t2 = fun()
# python_task = PythonOperator(
# task_id="in_dag_python_task",
# python_callable=run_task,
# )
# end = EmptyOperator(task_id="end")
# start >> t1 >> python_task >> end
# NOTE:
# NOTE: Runs the dag, but even if t2 is not in the dag dependency still it shows up in the dashboard, I couldn't understand the reason for it.
with DAG(
dag_id="dag_sample",
default_args=default_args,
description="A simple Airflow DAG",
schedule=timedelta(days=1), # Explicit schedule
catchup=False,
tags=["D1"],
) as dag:
start = EmptyOperator(task_id="start")
t1 = Foo().class_foo(dag)
t2 = Fun().class_fun(dag)
python_task = PythonOperator(
task_id="python_task", python_callable=run_task, op_args=["in_dag_python_task"]
)
end = EmptyOperator(task_id="end")
start >> t1 >> python_task >> end
```
Dear Team,
This is the minimal way, I can showcase my problem, here even if t2 is never a dag dependency it still shows up in dashboard.
By minimal, I meant in my use case, which is more complex has a similar scenario:
a) I have a dag blueprint that adds some core task groups run for every dag created by users.
b) My blueprint consists of a dag having multiple SUPER task groups lined to one another. Each alternating SUPER task group are similar in what functionality they perform. Each SUPER task group is a collection of:
- user_task groups (user functionalities)
- system_task groups (blueprint system functionalities, which change depending on context)
- some nominal tasks (stay as is, no matter the context)
c) I am facing problems with how the tasks inside the system_task groups are spawning inside the dag. Currently system_task groups are created by `methods` of `instances` of SystemTasks class, depending on context.
d) These functions are called by main dag, to get system tasks to be run alongside user tasks. In the current approach, dag is created by user, user_task_group is an input to my blueprint, where I simply chain user_task_group with blueprints system_task_groups.
e) The only problem I have is in the last SUPER task group, where the system_task group is spawning in the very beginning as a branch parallel to user_task group.
The code logic, I have gone through several times, but don't see a reason for this outlier branching, because it doesn't happen in the previous SUPER task group. So why only in the last one?
Hence I created the above dummy python script, where it seems even if I don't use a task inside the dependency, but it still shows inside the dag. There is no weird branching or linkage for the above simple python script like the one I got in my complex scenario, however, I still wanted to know is this intended or so?
GitHub link: https://github.com/apache/airflow/discussions/49502
----
This is an automatically sent email for commits@airflow.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@airflow.apache.org
Re: [D] DAG Tasks Creation [airflow]
Posted by "enchant3dmango (via GitHub)" <gi...@apache.org>.
GitHub user enchant3dmango added a comment to the discussion: DAG Tasks Creation
**TLDR: Yes, this is intended.**
Any task you instantiate inside a DAG context (```with DAG(...) as dag:```) is automatically added to the DAG, even if you don’t chain it to anything.
That’s why t2 shows up in the Airflow UI. (Doesn’t matter whether it’s linked or not — if it’s created in the DAG context, it’s part of the DAG.)
GitHub link: https://github.com/apache/airflow/discussions/49502#discussioncomment-12954376
----
This is an automatically sent email for commits@airflow.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@airflow.apache.org
Re: [D] DAG Tasks Creation [airflow]
Posted by "enchant3dmango (via GitHub)" <gi...@apache.org>.
GitHub user enchant3dmango added a comment to the discussion: DAG Tasks Creation
Check [this](https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/tasks.html#task-instances) out.
_Much in the same way that a DAG is instantiated into a [DAG Run](https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/dags.html#concepts-dag-run) each time it runs, the tasks under a DAG are instantiated into Task Instances._
GitHub link: https://github.com/apache/airflow/discussions/49502#discussioncomment-12954388
----
This is an automatically sent email for commits@airflow.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@airflow.apache.org
Re: [D] DAG Tasks Creation [airflow]
Posted by "Yuvraj-Dhepe (via GitHub)" <gi...@apache.org>.
GitHub user Yuvraj-Dhepe closed a discussion: DAG Tasks Creation
```python
from airflow.utils.task_group import TaskGroup
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.empty import EmptyOperator
from datetime import datetime, timedelta
import sys
def foo():
func_name = sys._getframe().f_code.co_name
python_task = PythonOperator(
task_id=func_name, python_callable=run_task, op_args=[func_name]
)
return python_task
def fun():
func_name = sys._getframe().f_code.co_name
python_task = PythonOperator(
task_id=func_name, python_callable=run_task, op_args=[func_name]
)
return python_task
class Foo:
def class_foo(self, dag):
func_name = sys._getframe().f_code.co_name
with TaskGroup(f"{func_name}", dag=dag) as task:
python_task = PythonOperator(
task_id=func_name, python_callable=run_task, op_args=[func_name]
)
# python_task
return task
class Fun:
def class_fun(self, dag):
func_name = sys._getframe().f_code.co_name
python_task = PythonOperator(
task_id=func_name, python_callable=run_task, op_args=[func_name]
)
# python_task
return python_task
def run_task(x):
print(f"Hello {x}")
default_args = {
"owner": "airflow",
"depends_on_past": False,
"start_date": datetime.now(),
"email_on_failure": False,
"email_on_retry": False,
"retries": 0,
}
# NOTE: Runs the dag, but even if t2 is not in the dag dependency still it shows up in the dashboard, I couldn't understand the reason for it.
# with DAG(
# dag_id="dag_sample",
# default_args=default_args,
# description="A simple Airflow DAG",
# schedule=timedelta(days=1), # Explicit schedule
# catchup=False,
# tags=["D1"],
# ) as dag:
# start = EmptyOperator(task_id="start")
# t1 = foo()
# t2 = fun()
# python_task = PythonOperator(
# task_id="in_dag_python_task",
# python_callable=run_task,
# )
# end = EmptyOperator(task_id="end")
# start >> t1 >> python_task >> end
# NOTE:
# NOTE: Runs the dag, but even if t2 is not in the dag dependency still it shows up in the dashboard, I couldn't understand the reason for it.
with DAG(
dag_id="dag_sample",
default_args=default_args,
description="A simple Airflow DAG",
schedule=timedelta(days=1), # Explicit schedule
catchup=False,
tags=["D1"],
) as dag:
start = EmptyOperator(task_id="start")
t1 = Foo().class_foo(dag)
t2 = Fun().class_fun(dag)
python_task = PythonOperator(
task_id="python_task", python_callable=run_task, op_args=["in_dag_python_task"]
)
end = EmptyOperator(task_id="end")
start >> t1 >> python_task >> end
```
Dear Team,
This is the minimal way, I can showcase my problem, here even if t2 is never a dag dependency it still shows up in dashboard.
By minimal, I meant in my use case, which is more complex has a similar scenario:
a) I have a dag blueprint that adds some core task groups run for every dag created by users.
b) My blueprint consists of a dag having multiple SUPER task groups lined to one another. Each alternating SUPER task group are similar in what functionality they perform. Each SUPER task group is a collection of:
- user_task groups (user functionalities)
- system_task groups (blueprint system functionalities, which change depending on context)
- some nominal tasks (stay as is, no matter the context)
c) I am facing problems with how the tasks inside the system_task groups are spawning inside the dag. Currently system_task groups are created by `methods` of `instances` of SystemTasks class, depending on context.
d) These functions are called by main dag, to get system tasks to be run alongside user tasks. In the current approach, dag is created by user, user_task_group is an input to my blueprint, where I simply chain user_task_group with blueprints system_task_groups.
e) The only problem I have is in the last SUPER task group, where the system_task group is spawning in the very beginning as a branch parallel to user_task group.
The code logic, I have gone through several times, but don't see a reason for this outlier branching, because it doesn't happen in the previous SUPER task group. So why only in the last one?
Hence I created the above dummy python script, where it seems even if I don't use a task inside the dependency, but it still shows inside the dag. There is no weird branching or linkage for the above simple python script like the one I got in my complex scenario, however, I still wanted to know is this intended or so?
GitHub link: https://github.com/apache/airflow/discussions/49502
----
This is an automatically sent email for commits@airflow.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@airflow.apache.org