Workflow Orchestration with Apache Airflow DAGs
Once you can write an Airflow DAG, the next problems are about design: files that slow the scheduler, branches that silently skip the final step, and dozens of near-identical pipelines. This guide runs four DAG design patterns on Airflow 3.3.2 and shows what each one fixes, with real output.
GloryTecks Team
Training & Placement Faculty
Last updated: 10 October 2026
Workflow orchestration with Apache Airflow DAGs starts simply: tasks, dependencies, a schedule. The Airflow tutorial covers that first DAG. The harder problems arrive later, when you have dozens of DAGs: the scheduler slows down, a branch leaves the final step silently skipped, and near-identical pipelines multiply by copy and paste.
This guide works through four design patterns that address those problems. Three of them run below on Airflow 3.3.2, using the official image and dag.test(), so you can see the behaviour rather than take it on trust.
import os, tempfile, textwrap, time
from datetime import datetime, timezone
import airflow
from airflow.configuration import conf
from airflow.dag_processing.dagbag import DagBag
from airflow.sdk import dag, task, task_group
UTC = timezone.utc
# Pattern 1: keep top-level code cheap. The same DAG, with slow work at module level or inside the task.
HEAVY = '''
import time, pandas
from datetime import datetime
from airflow.sdk import dag, task
time.sleep(2) # stands in for an API call made while building the DAG
TABLES = ["orders", "customers"]
@dag(schedule="@daily", start_date=datetime(2026, 1, 1), catchup=False)
def heavy():
@task
def load(table): return table
load.expand(table=TABLES)
heavy()
'''
LIGHT = '''
from datetime import datetime
from airflow.sdk import dag, task
@dag(schedule="@daily", start_date=datetime(2026, 1, 1), catchup=False)
def light():
@task
def tables():
import time; time.sleep(2) # the same slow call, now made when the task runs
return ["orders", "customers"]
@task
def load(table):
import pandas # heavy import only where it is used
return table
load.expand(table=tables())
light()
'''
# Pattern 2: branch, then join. The join's trigger rule decides whether it runs.
def branch_dag(join_rule):
@dag(dag_id=f"daily_load_{join_rule}", schedule=None, start_date=datetime(2026, 1, 1, tzinfo=UTC))
def daily_load():
@task.branch
def choose(logical_date=None):
return "full_reload" if logical_date.day == 1 else "incremental"
@task
def full_reload(): return "full"
@task
def incremental(): return "incremental"
@task(trigger_rule=join_rule)
def publish(): return "published"
choose() >> [full_reload(), incremental()] >> publish()
return daily_load()
branch_dags = {rule: branch_dag(rule) for rule in ("all_success", "none_failed_min_one_success")}
# Pattern 3: generate similar DAGs from configuration, with a task group per table
SOURCES = {"crm": ["accounts", "contacts"], "billing": ["invoices", "payments", "refunds"]}
def ingest_dag(source, tables):
@dag(dag_id=f"ingest_{source}", schedule="@daily", start_date=datetime(2026, 1, 1, tzinfo=UTC), catchup=False)
def ingest():
for table in tables:
@task_group(group_id=table)
def one_table():
@task
def extract(): return 1
@task
def load(rows): return rows
load(extract())
one_table()
return ingest()
generated = {source: ingest_dag(source, tables) for source, tables in SOURCES.items()}
if __name__ == "__main__": # Airflow re-imports this file to run tasks; report only once
print(f"P0 how often each DAG file is parsed (Airflow {airflow.__version__} defaults)")
print(" min_file_process_interval =", conf.get("dag_processor", "min_file_process_interval"), "seconds;",
"dag_file_processor_timeout =", conf.get("dag_processor", "dag_file_processor_timeout"), "seconds;",
"parsing_processes =", conf.get("dag_processor", "parsing_processes"))
import pandas # warm the import, so the timing below shows the top-level call itself
print("\nP1 time to parse one DAG file")
for name, source in [("slow work at top level", HEAVY), ("slow work inside tasks", LIGHT)]:
folder = tempfile.mkdtemp()
open(os.path.join(folder, "dag.py"), "w").write(textwrap.dedent(source))
start = time.perf_counter()
bag = DagBag(dag_folder=folder)
print(f" {name:<24} {time.perf_counter() - start:4.1f} seconds per parse | DAGs found: {list(bag.dags)}")
print("\nP2 branch then join, run for 2026-03-15 (an incremental day)")
for rule, d in branch_dags.items():
run = d.test(logical_date=datetime(2026, 3, 15, tzinfo=UTC))
states = {ti.task_id: str(ti.state) for ti in run.get_task_instances()}
print(f" join rule {rule:<29} full_reload={states['full_reload']}, incremental={states['incremental']},"
f" publish={states['publish']} | run {run.state}")
print("\nP3 DAGs generated from one configuration")
for source, d in generated.items():
print(f" {d.dag_id:<15} {len(d.tasks)} tasks: {sorted(t.task_id for t in d.tasks)}")P0 how often each DAG file is parsed (Airflow 3.3.2 defaults)
min_file_process_interval = 30 seconds; dag_file_processor_timeout = 50 seconds; parsing_processes = 2
P1 time to parse one DAG file
slow work at top level 2.0 seconds per parse | DAGs found: ['heavy']
slow work inside tasks 0.0 seconds per parse | DAGs found: ['light']
P2 branch then join, run for 2026-03-15 (an incremental day)
join rule all_success full_reload=skipped, incremental=success, publish=skipped | run success
join rule none_failed_min_one_success full_reload=skipped, incremental=success, publish=success | run success
P3 DAGs generated from one configuration
ingest_crm 4 tasks: ['accounts.extract', 'accounts.load', 'contacts.extract', 'contacts.load']
ingest_billing 6 tasks: ['invoices.extract', 'invoices.load', 'payments.extract', 'payments.load', 'refunds.extract', 'refunds.load']Pattern 1: keep top-level code cheap
Airflow does not read a DAG file once. Everything outside a task's execution runs every time the file is parsed, and in Airflow 3.3.2 the default is to re-parse each file every 30 seconds, using 2 parsing processes, with a 50-second timeout per file (P0). The documentation is explicit: no database access, heavy computation or network calls at the top level, and expensive imports belong inside the callables that use them (Airflow: best practices).
The measurement shows why. The same DAG took 2.0 seconds to parse when a slow call ran at the top level, and effectively no time when that call moved into a task (P1). Two seconds sounds small, but it is paid on every parse of every such file. As a rough estimate with the defaults above, two parsing processes can absorb at most about 30 such files before parsing falls behind its 30-second cycle, delaying changes and, in the worst case, timing files out.
The pattern: build the DAG structure from constants, environment variables or a configuration file in the DAG folder, and do anything slow, such as listing tables from a database or calling an API, inside a task. If the list of items comes from a slow source at run time, return it from a task and use dynamic task mapping over it, as the light version does.
Pattern 2: join branches with the right trigger rule
Branching is common: a full reload on the first of the month, an incremental load otherwise, then a shared final step. The trap is that skipped tasks cascade through the default trigger rule, all_success (Airflow: Dags).
In P2, on an incremental day, the full_reload branch was skipped, as intended. With the default rule, publish was skipped too, and the run still finished as success. Nothing was published and nothing turned red. With none_failed_min_one_success, the rule the documentation recommends for this case, publish ran.
The pattern: every task that joins branches needs an explicit trigger rule, usually none_failed_min_one_success. Add a check in your DAG tests that the final task of every run actually succeeded, not just the run.
Pattern 3: generate similar DAGs from configuration
When ten sources need the same extract-and-load shape, copying a DAG file ten times creates ten things to fix. Generating DAGs from configuration keeps one definition. In P3, one dictionary produced two DAGs, ingest_crm and ingest_billing, each with a task group per table, so task IDs read as invoices.extract and invoices.load.
The documentation adds the rules that keep this safe: read configuration from environment variables or a file in the DAG folder rather than Airflow Variables at the top level, generate tasks in a consistent order, and consider get_parsing_context() when one file generates many DAGs, so a running task does not rebuild them all (Airflow: dynamic DAG generation). Use generation for DAGs that differ only by configuration; when the number of tasks depends on data at run time, use dynamic task mapping instead.
Pattern 4: connect DAGs with assets, not sensors
A large DAG that does everything is hard to own and rerun. Splitting it into a producer and consumers is better, but the consumers then need to know when the producer has finished. In Airflow 3, a producer task declares an asset in its outlets, and a consumer DAG is scheduled on that asset. The asset is marked updated only when the producer task succeeds, so a failed or skipped producer does not start the consumers (Airflow: asset scheduling).
from datetime import datetime, timezone
from airflow.sdk import Asset, dag, task
orders_table = Asset("warehouse://analytics/orders")
@dag(schedule="@daily", start_date=datetime(2026, 1, 1, tzinfo=timezone.utc), catchup=False)
def load_orders():
@task(outlets=[orders_table])
def load():
... # write the day's orders
load()
@dag(schedule=[orders_table], start_date=datetime(2026, 1, 1, tzinfo=timezone.utc), catchup=False)
def refresh_revenue_report():
@task
def refresh():
... # runs only after load_orders has updated the asset
refresh()
load_orders()
refresh_revenue_report()Compared with a sensor that polls for the producer's result, the consumer uses no worker slot while it waits, and the dependency is visible in the UI.
A review checklist for DAGs
- No slow work at the top level: no API calls, database queries or heavy imports outside tasks.
- Explicit trigger rules on every join after a branch.
- Final-step assertions in DAG tests, not just run status.
- One generated definition for pipelines that differ only by configuration.
- Assets between DAGs instead of sensors where both DAGs live in Airflow.
- Idempotent tasks, since Airflow retries; the Airflow interview questions show safe reruns and failure handling.
For choosing between Airflow and other orchestrators, see Airflow vs Dagster vs Prefect.
Why should I avoid top-level code in Airflow DAG files?
Why was my Airflow task skipped after a branch?
How do I create many similar DAGs in Airflow?
How do I trigger one Airflow DAG after another finishes?
Airflow orchestration and pipeline design are taught in the Data Engineering course.
Sources
- Best Practices (Apache Airflow documentation) — Apache Software Foundation
- Dags (Apache Airflow documentation) — Apache Software Foundation
- Dynamic Dag Generation (Apache Airflow documentation) — Apache Software Foundation
- Asset-aware scheduling (Apache Airflow documentation) — Apache Software Foundation
GloryTecks Team
Training & Placement Faculty
The GloryTecks faculty team mentors learners across Data Science, AI and Data Engineering, with a focus on real-time projects and placement readiness in Hyderabad.
Want to master Data Engineering?
Join GloryTecks’ expert-led Data Engineering training in Hyderabad with real-time projects and 100% placement support.
Related articles
Apache Spark Explained for Beginners
A beginner's guide to Apache Spark: what it is for, how the driver, executors and cluster manager fit together, jobs, stages and tasks, lazy transformations and actions demonstrated on three million New York taxi trips, partitions, installing PySpark, and the mistakes beginners make.
Apache Iceberg vs Delta Lake vs Hudi
A comparison of the three open table formats: the problems they all solve on a data lake, Iceberg's hidden partitioning and partition evolution, Delta Lake's transaction log and Spark roots, Hudi's copy-on-write and merge-on-read tables and incremental queries, interoperability through UniForm, and a choice guide.
Data Engineering Roadmap 2026: A Complete Guide
A stage-by-stage data engineering roadmap for 2026: what the role involves according to current certification guides, then seven stages, SQL, Python, pipelines and modelling, orchestration and quality, Spark and big data, streaming, and cloud platforms, each with a checkpoint, and detours to avoid.