如何在Airflow中实现所有任务执行完成后再执行指定任务

Hey there! Let's break down how to add those two new tasks that wait for all your 10 existing tasks to finish. I'll cover a couple of clean, maintainable approaches depending on your Airflow version:

Approach 1: Use a Dummy Operator as a "Completion Trigger" (Works for All Airflow Versions) #

This is the most straightforward method, especially if you're on Airflow 1.x or prefer simplicity. We'll create a dummy task that acts as a gatekeeper—all existing tasks point to it, and your new tasks only run after this dummy completes.

Here's a code example:

from airflow import DAG
from airflow.operators.dummy import DummyOperator
from airflow.operators.python import PythonOperator
from datetime import datetime

def existing_task_func():
    # Your existing task logic here
    pass

def new_task_1_func():
    # Logic for first new task
    pass

def new_task_2_func():
    # Logic for second new task
    pass

with DAG(
    dag_id='your_dag_id',
    start_date=datetime(2024, 1, 1),
    schedule_interval='@daily'
) as dag:
    # Define all existing tasks (example with 10 tasks)
    existing_tasks = [
        PythonOperator(task_id=f'existing_task_{i}', python_callable=existing_task_func)
        for i in range(10)
    ]

    # Create a dummy task that waits for all existing tasks
    all_existing_completed = DummyOperator(task_id='all_existing_tasks_completed')

    # Set dependencies: all existing tasks >> dummy task
    for task in existing_tasks:
        task >> all_existing_completed

    # Define new tasks, dependent on the dummy task
    new_task_1 = PythonOperator(task_id='new_task_1', python_callable=new_task_1_func)
    new_task_2 = PythonOperator(task_id='new_task_2', python_callable=new_task_2_func)

    all_existing_completed >> [new_task_1, new_task_2]

    # Optional: If new_task_2 needs to run after new_task_1, add this line
    # new_task_1 >> new_task_2

Approach 2: Use Task Groups (Airflow 2.0+) #

If you're using Airflow 2.0 or later, Task Groups are a cleaner way to organize your existing tasks. You can group all 10 tasks into a single logical group, then set your new tasks to depend on the entire group.

Example code:

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.utils.task_group import TaskGroup
from datetime import datetime

def existing_task_func():
    # Your existing task logic here
    pass

def new_task_1_func():
    # Logic for first new task
    pass

def new_task_2_func():
    # Logic for second new task
    pass

with DAG(
    dag_id='your_dag_id',
    start_date=datetime(2024, 1, 1),
    schedule_interval='@daily'
) as dag:
    # Group all existing tasks into a TaskGroup
    with TaskGroup(group_id='existing_tasks_group') as existing_tasks_group:
        for i in range(10):
            PythonOperator(task_id=f'existing_task_{i}', python_callable=existing_task_func)

    # Define new tasks, dependent on the entire TaskGroup
    new_task_1 = PythonOperator(task_id='new_task_1', python_callable=new_task_1_func)
    new_task_2 = PythonOperator(task_id='new_task_2', python_callable=new_task_2_func)

    existing_tasks_group >> [new_task_1, new_task_2]

    # Optional: Add dependency between new tasks if needed
    # new_task_1 >> new_task_2

Quick Note on Direct Dependencies (Not Recommended) #

You could technically set each new task to depend on all 10 existing tasks directly, like this:

new_task_1.set_upstream(existing_tasks)
new_task_2.set_upstream(existing_tasks)

But this gets messy fast—if you add or remove existing tasks later, you'll have to update these dependencies every time. The first two approaches are far more maintainable.

Key Tips #

• Make sure your existing tasks' internal dependencies are already correctly configured (they should all reach a completed state before the dummy/group triggers the new tasks).
• If your two new tasks don't need to run in sequence, they'll execute in parallel once the existing tasks finish—perfect if they're independent!

内容的提问来源于stack exchange，提问作者Shanmukh S