Designing the ETL Pipeline
A classic ETL DAG breaks work into three clear stages, each as its own task or group of tasks: extract pulls raw data from a source system (an API, a database, a file drop) into a staging location; transform cleans, joins, and reshapes that raw data into the target schema; and load writes the transformed data into its final destination, such as a data warehouse table. Designing the DAG means deciding the granularity of these stages, for example splitting extract into extract_customers and extract_orders as parallel tasks if the sources are independent, which lets Airflow run them concurrently rather than serially, cutting overall pipeline runtime.
Cricket analogy: Splitting extract_customers and extract_orders to run in parallel is like two support staff independently checking the pitch report and the weather forecast at the same time before a toss, rather than one person doing both sequentially.
Implementing Extract, Transform, Load Tasks
Using the TaskFlow API, each stage becomes a plain Python function decorated with @task, and passing data between them (small metadata, not the full dataset) happens simply by calling one function with the return value of another, which Airflow automatically wires into XCom under the hood. For example, extract() might query an API and write results to a staging path in S3 or a local parquet file, returning just that path; transform(path) reads the staged file, applies pandas or SQL transformations, and writes a cleaned version to a new staging path; and load(path) reads the cleaned file and writes it into the warehouse table using an idempotent MERGE or DELETE+INSERT.
Cricket analogy: Passing just a file path between extract and transform, rather than the whole dataset, is like a scorer passing the match scorecard reference number to the analyst rather than physically carrying every ball-by-ball sheet across the ground.
from airflow.decorators import dag, task
from pendulum import datetime
import pandas as pd
@dag(
dag_id="orders_etl",
schedule="@daily",
start_date=datetime(2026, 1, 1),
catchup=False,
default_args={"retries": 3, "retry_delay": 180},
tags=["etl", "orders"],
)
def orders_etl():
@task
def extract(ds=None) -> str:
# Pull raw orders for the logical date from the source API
raw_path = f"/data/staging/orders_raw_{ds}.parquet"
# ... fetch and write raw_path ...
return raw_path
@task
def transform(raw_path: str, ds=None) -> str:
df = pd.read_parquet(raw_path)
df = df.dropna(subset=["order_id", "customer_id"])
df["total_amount"] = df["quantity"] * df["unit_price"]
clean_path = f"/data/staging/orders_clean_{ds}.parquet"
df.to_parquet(clean_path)
return clean_path
@task
def load(clean_path: str, ds=None) -> None:
# DELETE FROM orders_fact WHERE partition_date = '{ds}';
# COPY INTO orders_fact FROM '{clean_path}';
pass
load(transform(extract()))
orders_etl()Handling Failures and Retries
Each task can define retries and retry_delay (either per-task or via default_args for the whole DAG) so a transient failure, like a momentary API timeout, doesn't fail the entire pipeline run on the first attempt. For failures that exhaust retries, an on_failure_callback function can send an alert to Slack or PagerDuty with the task instance's context (task_id, execution date, log URL), giving on-call engineers immediate, actionable context instead of requiring them to dig through the Airflow UI first. It's also good practice to set sla or use on_execute_timeout/execution_timeout on long-running extract or transform tasks so a hung API call doesn't silently block downstream tasks for hours.
Cricket analogy: Giving a task retries with a delay is like a bowler getting a re-run of a no-ball delivery after a brief reset, rather than the innings ending outright on the first miscue.
Testing the DAG Locally
Before deploying, run airflow dags list-import-errors to catch syntax or import problems, then use airflow tasks test orders_etl extract 2026-07-10 to execute a single task in isolation with full logging, which is invaluable for debugging one broken stage without rerunning the whole pipeline. For a full pipeline dry run, airflow dags test orders_etl 2026-07-10 executes every task for that logical date end-to-end locally, and pointing the DAG's staging paths at a temp or sandbox location (rather than production tables) during this test avoids polluting real data while you validate the transform logic and load idempotency.
Cricket analogy: Testing one task in isolation with airflow tasks test is like a bowler practicing a specific delivery alone in the nets rather than needing a full 11-a-side match just to check one skill.
Use airflow tasks test <dag_id> <task_id> <date> to debug a single failing task with full log output printed to your terminal, without needing to trigger the whole DAG or wait for the scheduler.
Never pass a full pandas DataFrame or raw file bytes as a return value between TaskFlow tasks. Airflow serializes XCom values (often via pickling or JSON) into the metadata database, and large payloads will bloat the database and can crash the webserver's UI when it tries to render them.
- Structure ETL DAGs into clear extract, transform, and load stages, parallelizing independent extracts where possible.
- Use the TaskFlow API so each stage is a plain Python function connected via lightweight return values.
- Pass only small metadata (file paths, row counts) between tasks; write actual data to external staging storage.
- Set retries and retry_delay to absorb transient failures without failing the whole pipeline immediately.
- Use on_failure_callback to send actionable alerts (task, date, log URL) when retries are exhausted.
- Use execution_timeout to prevent a hung task from silently blocking downstream tasks indefinitely.
- Test locally with
airflow tasks testfor single tasks andairflow dags testfor full end-to-end dry runs before deploying.
Practice what you learned
1. Why might extract_customers and extract_orders be modeled as two separate parallel tasks rather than one combined task?
2. In the TaskFlow API, what should a task like extract() typically return when handing off data to the next task?
3. What is the purpose of an on_failure_callback?
4. Which command lets you test a single task in isolation with full log output in your terminal?
5. Why is an idempotent load pattern (like DELETE+INSERT) important in the load task of an ETL DAG?
Was this page helpful?
You May Also Like
Airflow Best Practices
Practical guidelines for writing reliable, maintainable Airflow DAGs, covering idempotency, DAG design, dependency management, and testing.
Airflow Quick Reference
A condensed cheat sheet of essential Airflow CLI commands, common operators/sensors, scheduling syntax, and key configuration settings.
Airflow vs Prefect vs Dagster
A comparison of three popular Python workflow orchestrators, covering their core philosophies, execution models, and when to choose each.
Related Reading
Related Study Notes in Programming
Browse all study notesApache Spark Study Notes
Programming · 30 topics
ProgrammingApache Flink Study Notes
Programming · 30 topics
ProgrammingHadoop Study Notes
Programming · 30 topics
ProgrammingSnowflake Study Notes
Programming · 30 topics
Programmingdbt (Data Build Tool) Study Notes
Programming · 30 topics
ProgrammingPowerShell Study Notes
Programming · 30 topics