Workflows often need to handle repetitive patterns (e.g., processing 100 files) or branching logic (e.g., run task B if condition is met, otherwise task C). Writing each task manually can be inefficient.

In this lesson, you’ll learn how to:

1. Task Dependencies in Airflow

  • Dependencies define the order of execution.

  • Two main operators:

    • task1 >> task2 → task1 runs before task2.

    • task1 << task2 → task2 runs before task1.

  • You can chain multiple tasks:

    task1 >> [task2, task3] >> task4

    This means:

    • task2 and task3 both depend on task1.

    • task4 depends on both task2 and task3.

2. Dynamic Task Generation

Instead of writing repetitive code, you can dynamically create tasks in a loop.

Example: Processing multiple files.

from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime

files = ["data1.csv", "data2.csv", "data3.csv"]

with DAG(
    dag_id="dynamic_dag_example",
    start_date=datetime(2025, 1, 1),
    schedule_interval="@daily",
    catchup=False
) as dag:

    start = BashOperator(
        task_id="start",
        bash_command="echo 'Starting pipeline'"
    )

    for f in files:
        process = BashOperator(
            task_id=f"process_{f}",
            bash_command=f"echo 'Processing {f}'"
        )
        start >> process

Here:

  • Tasks are generated dynamically for each file.

  • If you add more files, new tasks are created automatically.

3. Conditional Branching with BranchPythonOperator

Sometimes workflows need decisions:

  • If today is Monday → run weekly report.

  • Otherwise → skip.

Example:

from airflow.operators.python import BranchPythonOperator, PythonOperator

def choose_branch(**context):
    from datetime import datetime
    if datetime.today().weekday() == 0:  # Monday
        return "weekly_report"
    else:
        return "skip_task"

branch = BranchPythonOperator(
    task_id="branching",
    python_callable=choose_branch
)

weekly_report = PythonOperator(
    task_id="weekly_report",
    python_callable=lambda: print("Running weekly report")
)

skip_task = PythonOperator(
    task_id="skip_task",
    python_callable=lambda: print("No report today")
)

branch >> [weekly_report, skip_task]

Here:

  • The BranchPythonOperator decides which path to take.

  • Only the selected task will run.

Best Practices

  • Use loops for repetitive tasks, but keep DAGs readable.

  • Avoid creating too many tasks dynamically (can slow down the scheduler).

  • Use branching for conditions instead of putting logic inside a single task.

  • Document dependencies clearly for maintainability.

Lesson Summary

  • Task dependencies ensure workflows run in the correct order.

  • Dynamic DAGs let you scale workflows without writing repetitive code.

  • Branching allows conditional paths inside a DAG.

  • With these tools, Airflow can handle complex, real-world pipelines.

PreviousNext