Suppose a sales pipeline must process data from two ERP systems, with the source determined by the DAG run’s data interval. Branching selects the appropriate execution path. If both systems need nearly identical fetch and clean steps, a PythonOperator can select code paths inside those tasks without changing the DAG structure. section 6.2.1
Internal branching becomes less useful when the systems require different chains of tasks: the DAG view cannot show which code path ran, and general PythonOperators may replace more specialized operators. Define separate task paths when those differences matter operationally. section 6.2.1
Choose an ERP path in the DAG
Create an old fetch-to-clean path and a new fetch-to-clean path. Put a BranchPythonOperator before them; its callable returns "fetch_sales_old" when data_interval_start precedes ERP_CHANGE_DATE, or "fetch_sales_new" otherwise. Connect the branch task to both fetch tasks. It may also return a list of downstream task IDs to select multiple tasks. section 6.2.2
Listing 6.11 Adding the branching condition function
def _pick_erp_system(**context):
if context["data_interval_start "] < ERP_CHANGE_DATE:
return "fetch_sales_old"
else:
return "fetch_sales_new"
pick_erp_system = BranchPythonOperator(
task_id="pick_erp_system",
python_callable=_pick_erp_system,
)
pick_erp_system >> [fetch_sales_old, fetch_sales_new]
The branching callable and its downstream dependencies show how a run selects the old or new fetch path. section 6.2.2
Take care where the paths rejoin. The unchosen path is skipped, so a join task using the default all_success trigger rule is also skipped: not all its upstream tasks succeeded. Set trigger_rule="none_failed" on the joining task. Alternatively, place an EmptyOperator with that rule between both clean tasks and the existing downstream task; this makes the join explicit without changing the downstream task’s rule. section 6.2.2
Listing 6.15 Adding an empty join task for clarity
from airflow.providers.standard.operators.empty import EmptyOperator
join_branch = EmptyOperator(
task_id="join_erp_branch",
trigger_rule="none_failed"
)
[clean_sales_old, clean_sales_new] >> join_branch
join_branch >> join_datasets
The empty join task collects both paths and applies none_failed before the workflow continues. section 6.2.2
