Il 15 settembre 2026, tutti gli ambienti Cloud Composer 1 e Cloud Composer 2 versione 2.0.x raggiungeranno la fine del ciclo di vita pianificata e non potrai più utilizzarli. Ti consigliamo di pianificare la migrazione a Cloud Composer 3.
Questa pagina descrive come raggruppare le attività nelle pipeline Airflow
utilizzando i seguenti pattern di progettazione:
Raggruppamento delle attività nel grafico DAG.
Attivazione dei DAG secondari da un DAG principale.
Raggruppamento delle attività con l'operatore TaskGroup.
Raggruppare le attività nel grafico DAG
Per raggruppare le attività in determinate fasi della pipeline, puoi utilizzare le relazioni tra le attività nel file DAG.
Considera l'esempio seguente:
Figura 1. Le attività possono essere raggruppate in un DAG
Airflow (fai clic per ingrandire)
In questo flusso di lavoro, le attività op-1 e op-2 vengono eseguite insieme dopo l'attività
iniziale start. Puoi farlo raggruppando le attività con l'istruzione
start >> [task_1, task_2].
Il seguente esempio fornisce un'implementazione completa di questo DAG:
fromairflowimportDAGfromairflow.operators.bashimportBashOperatorfromairflow.operators.dummyimportDummyOperatorfromairflow.utils.datesimportdays_agoDAG_NAME="all_tasks_in_one_dag"args={"owner":"airflow","start_date":days_ago(1),"schedule_interval":"@once"}withDAG(dag_id=DAG_NAME,default_args=args)asdag:start=DummyOperator(task_id="start")task_1=BashOperator(task_id="op-1",bash_command=":",dag=dag)task_2=BashOperator(task_id="op-2",bash_command=":",dag=dag)some_other_task=DummyOperator(task_id="some-other-task")task_3=BashOperator(task_id="op-3",bash_command=":",dag=dag)task_4=BashOperator(task_id="op-4",bash_command=":",dag=dag)end=DummyOperator(task_id="end")start >> [task_1,task_2] >> some_other_task >> [task_3,task_4] >> end
Figura 2. I DAG possono essere attivati dall'interno di un DAG con
TriggerDagRunOperator (fai clic per ingrandire)
In questo flusso di lavoro, i blocchi dag_1 e dag_2 rappresentano una serie di attività
raggruppate in un DAG separato nell'ambiente
Cloud Composer.
L'implementazione di questo flusso di lavoro richiede due file DAG separati.
Il file DAG di controllo ha il seguente aspetto:
fromairflowimportDAGfromairflow.operators.dummyimportDummyOperatorfromairflow.operators.trigger_dagrunimportTriggerDagRunOperatorfromairflow.utils.datesimportdays_agowithDAG(dag_id="controller_dag_to_trigger_other_dags",default_args={"owner":"airflow"},start_date=days_ago(1),schedule_interval="@once",)asdag:start=DummyOperator(task_id="start")trigger_1=TriggerDagRunOperator(task_id="dag_1",trigger_dag_id="dag-to-trigger",# Ensure this equals the dag_id of the DAG to triggerconf={"message":"Hello World"},)trigger_2=TriggerDagRunOperator(task_id="dag_2",trigger_dag_id="dag-to-trigger",# Ensure this equals the dag_id of the DAG to triggerconf={"message":"Hello World"},)some_other_task=DummyOperator(task_id="some-other-task")end=DummyOperator(task_id="end")start >> trigger_1 >> some_other_task >> trigger_2 >> end
L'implementazione del DAG secondario, attivato dal DAG di controllo, è simile alla seguente:
Raggruppamento delle attività con l'operatore TaskGroup
Puoi utilizzare l'operatore
TaskGroup per raggruppare le attività
nel DAG. Le attività definite all'interno di un blocco TaskGroup fanno ancora parte
del DAG principale.
Considera l'esempio seguente:
Figura 3. Le attività possono essere raggruppate visivamente nell'interfaccia utente con l'operatore TaskGroup (fai clic per ingrandire)
Le attività op-1 e op-2 sono raggruppate in un blocco con ID
taskgroup_1. Un'implementazione di questo flusso di lavoro è simile al seguente codice:
fromairflow.models.dagimportDAGfromairflow.operators.bashimportBashOperatorfromairflow.operators.dummyimportDummyOperatorfromairflow.utils.datesimportdays_agofromairflow.utils.task_groupimportTaskGroupwithDAG(dag_id="taskgroup_example",start_date=days_ago(1))asdag:start=DummyOperator(task_id="start")withTaskGroup("taskgroup_1",tooltip="task group #1")assection_1:task_1=BashOperator(task_id="op-1",bash_command=":")task_2=BashOperator(task_id="op-2",bash_command=":")withTaskGroup("taskgroup_2",tooltip="task group #2")assection_2:task_3=BashOperator(task_id="op-3",bash_command=":")task_4=BashOperator(task_id="op-4",bash_command=":")some_other_task=DummyOperator(task_id="some-other-task")end=DummyOperator(task_id="end")start >> section_1 >> some_other_task >> section_2 >> end
[[["Facile da capire","easyToUnderstand","thumb-up"],["Il problema è stato risolto","solvedMyProblem","thumb-up"],["Altra","otherUp","thumb-up"]],[["Difficile da capire","hardToUnderstand","thumb-down"],["Informazioni o codice di esempio errati","incorrectInformationOrSampleCode","thumb-down"],["Mancano le informazioni o gli esempi di cui ho bisogno","missingTheInformationSamplesINeed","thumb-down"],["Problema di traduzione","translationIssue","thumb-down"],["Altra","otherDown","thumb-down"]],["Ultimo aggiornamento 2025-08-26 UTC."],[[["\u003cp\u003eThis document outlines methods for grouping tasks within Airflow pipelines, covering approaches like structuring relationships in the DAG graph, triggering child DAGs from a parent DAG, and utilizing the \u003ccode\u003eTaskGroup\u003c/code\u003e operator.\u003c/p\u003e\n"],["\u003cp\u003eGrouping tasks directly in the DAG graph is achieved by defining relationships between tasks, demonstrated with the syntax \u003ccode\u003estart >> [task_1, task_2]\u003c/code\u003e, which executes \u003ccode\u003etask_1\u003c/code\u003e and \u003ccode\u003etask_2\u003c/code\u003e concurrently after \u003ccode\u003estart\u003c/code\u003e.\u003c/p\u003e\n"],["\u003cp\u003eParent DAGs can trigger child DAGs using the \u003ccode\u003eTriggerDagRunOperator\u003c/code\u003e, requiring the \u003ccode\u003etrigger_dag_id\u003c/code\u003e to match the \u003ccode\u003edag_id\u003c/code\u003e of the child DAG.\u003c/p\u003e\n"],["\u003cp\u003eThe \u003ccode\u003eTaskGroup\u003c/code\u003e operator allows for grouping tasks within a DAG, which provides a visual organization in the Airflow UI and simplifies complex workflows.\u003c/p\u003e\n"],["\u003cp\u003eIt is recommended to avoid using SubDAGs for grouping tasks due to performance and functional issues; the document presents superior alternative methods for structuring workflows.\u003c/p\u003e\n"]]],[],null,["\u003cbr /\u003e\n\n\u003cbr /\u003e\n\n\n[Cloud Composer 3](/composer/docs/composer-3/group-tasks-inside-dags \"View this page for Cloud Composer 3\") \\| **Cloud Composer 2** \\| [Cloud Composer 1](/composer/docs/composer-1/group-tasks-inside-dags \"View this page for Cloud Composer 1\")\n\n\u003cbr /\u003e\n\n\u003cbr /\u003e\n\n\u003cbr /\u003e\n\n\u003cbr /\u003e\n\n\u003cbr /\u003e\n\n\u003cbr /\u003e\n\nThis page describes how you can group tasks in your Airflow pipelines\nusing the following design patterns:\n\n- Grouping tasks in the DAG graph.\n- Triggering children DAGs from a parent DAG.\n- Grouping tasks with the `TaskGroup` operator.\n\n| **Important:** Airflow provides [SubDAGs](https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/dags.html#subdags) to address repeating tasks. Despite being a common design pattern for grouping tasks together, SubDAGs often cause performance and functional issues, and is deprecated in Airflow. We recommend to **avoid using SubDAGs to group tasks together** in your workflow and prefer one of the alternative approaches described in this page.\n\nGroup tasks in the DAG graph\n\nTo group tasks in certain phases of your pipeline, you can use relationships\nbetween the tasks in your DAG file.\n\nConsider the following example:\n[](/static/composer/docs/images/workflow-group-dags.png) **Figure 1.** Tasks can be grouped together in an Airflow DAG (click to enlarge)\n\nIn this workflow, tasks `op-1` and `op-2` run together after the initial\ntask `start`. You can achieve this by grouping tasks together with the statement\n`start \u003e\u003e [task_1, task_2]`.\n\nThe following example provides a complete implementation of this DAG:\n\n\n from airflow import DAG\n from airflow.operators.bash import BashOperator\n from airflow.operators.dummy import DummyOperator\n from airflow.utils.dates import days_ago\n\n DAG_NAME = \"all_tasks_in_one_dag\"\n\n args = {\"owner\": \"airflow\", \"start_date\": days_ago(1), \"schedule_interval\": \"@once\"}\n\n with DAG(dag_id=DAG_NAME, default_args=args) as dag:\n start = DummyOperator(task_id=\"start\")\n\n task_1 = BashOperator(task_id=\"op-1\", bash_command=\":\", dag=dag)\n\n task_2 = BashOperator(task_id=\"op-2\", bash_command=\":\", dag=dag)\n\n some_other_task = DummyOperator(task_id=\"some-other-task\")\n\n task_3 = BashOperator(task_id=\"op-3\", bash_command=\":\", dag=dag)\n\n task_4 = BashOperator(task_id=\"op-4\", bash_command=\":\", dag=dag)\n\n end = DummyOperator(task_id=\"end\")\n\n start \u003e\u003e [task_1, task_2] \u003e\u003e some_other_task \u003e\u003e [task_3, task_4] \u003e\u003e end\n\n\u003cbr /\u003e\n\n\nTrigger children DAGs from a parent DAG\n\nYou can trigger one DAG from another DAG with the\n[`TriggerDagRunOperator` operator](https://airflow.apache.org/docs/apache-airflow/stable/_api/airflow/operators/trigger_dagrun/).\n\nConsider the following example:\n[](/static/composer/docs/images/workflow-trigger-dags.png) **Figure 2.** DAGs can be triggered from within a DAG with the TriggerDagRunOperator (click to enlarge)\n\nIn this workflow, the blocks `dag_1` and `dag_2` represent a series of tasks\nthat are grouped together in a separate DAG in the Cloud Composer\nenvironment.\n\nThe implementation of this workflow requires two separate DAG files.\nThe controlling DAG file looks like the following:\n\n\n from airflow import DAG\n from airflow.operators.dummy import DummyOperator\n from airflow.operators.trigger_dagrun import TriggerDagRunOperator\n from airflow.utils.dates import days_ago\n\n\n with DAG(\n dag_id=\"controller_dag_to_trigger_other_dags\",\n default_args={\"owner\": \"airflow\"},\n start_date=days_ago(1),\n schedule_interval=\"@once\",\n ) as dag:\n start = DummyOperator(task_id=\"start\")\n\n trigger_1 = TriggerDagRunOperator(\n task_id=\"dag_1\",\n trigger_dag_id=\"dag-to-trigger\", # Ensure this equals the dag_id of the DAG to trigger\n conf={\"message\": \"Hello World\"},\n )\n trigger_2 = TriggerDagRunOperator(\n task_id=\"dag_2\",\n trigger_dag_id=\"dag-to-trigger\", # Ensure this equals the dag_id of the DAG to trigger\n conf={\"message\": \"Hello World\"},\n )\n\n some_other_task = DummyOperator(task_id=\"some-other-task\")\n\n end = DummyOperator(task_id=\"end\")\n\n start \u003e\u003e trigger_1 \u003e\u003e some_other_task \u003e\u003e trigger_2 \u003e\u003e end\n\n\u003cbr /\u003e\n\n\n| **Note:** The value for `trigger_dag_id` inside `TriggerDagRunOperator` must match the `dag_id` value of the DAG you want to trigger.\n\nThe implementation of the child DAG, which is triggered by the controlling\nDAG, looks like the following:\n\n\n from airflow import DAG\n from airflow.operators.dummy import DummyOperator\n from airflow.utils.dates import days_ago\n\n DAG_NAME = \"dag-to-trigger\"\n\n args = {\"owner\": \"airflow\", \"start_date\": days_ago(1), \"schedule_interval\": \"None\"}\n\n with DAG(dag_id=DAG_NAME, default_args=args) as dag:\n dag_task = DummyOperator(task_id=\"dag-task\")\n\n\u003cbr /\u003e\n\n\nYou must [upload both DAG files](/composer/docs/composer-2/manage-dags#add)\nin your Cloud Composer environment for the DAG to work.\n\nGrouping tasks with the TaskGroup operator\n\nYou can use the\n[`TaskGroup` operator](https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/dags.html#taskgroups) to group tasks\ntogether in your DAG. Tasks defined within a `TaskGroup` block are still part\nof the main DAG.\n\nConsider the following example:\n[](/static/composer/docs/images/workflow-taskgroup-dag.png) **Figure 3.** Tasks can be visually grouped together in the UI with the TaskGroup operator (click to enlarge)\n\nThe tasks `op-1` and `op-2` are grouped together in a block with ID\n`taskgroup_1`. An implementation of this workflow looks like the following code: \n\n from airflow.models.dag import DAG\n from airflow.operators.bash import BashOperator\n from airflow.operators.dummy import DummyOperator\n from airflow.utils.dates import days_ago\n from airflow.utils.task_group import TaskGroup\n\n with DAG(dag_id=\"taskgroup_example\", start_date=days_ago(1)) as dag:\n start = DummyOperator(task_id=\"start\")\n\n with TaskGroup(\"taskgroup_1\", tooltip=\"task group #1\") as section_1:\n task_1 = BashOperator(task_id=\"op-1\", bash_command=\":\")\n task_2 = BashOperator(task_id=\"op-2\", bash_command=\":\")\n\n with TaskGroup(\"taskgroup_2\", tooltip=\"task group #2\") as section_2:\n task_3 = BashOperator(task_id=\"op-3\", bash_command=\":\")\n task_4 = BashOperator(task_id=\"op-4\", bash_command=\":\")\n\n some_other_task = DummyOperator(task_id=\"some-other-task\")\n\n end = DummyOperator(task_id=\"end\")\n\n start \u003e\u003e section_1 \u003e\u003e some_other_task \u003e\u003e section_2 \u003e\u003e end\n\nWhat's next\n\n- [Write DAGs](/composer/docs/composer-2/write-dags)\n- [Test DAGs](/composer/docs/composer-2/test-dags)"]]