Building Your First Airflow DAG for ETL Using the TaskFlow API
In this article, let us build our first Airflow DAG for an ETL use case using the TaskFlow API. We will read a daily CSV file, validate and transform the records, and load them into Postgres. This is a small example, but it covers the structure we normally need in a real pipeline: extract, transform, load, retries, connections, and a safe way to rerun the job.
I prefer the TaskFlow API for a new Python-based DAG because the dependencies are easier to read than a long list of PythonOperator objects. It also removes some boilerplate around XCom. We still need to understand what Airflow is doing behind the scenes, especially when data is passed between tasks.
What We Are Building
Assume another system places a file such as /opt/airflow/data/orders_2026-10-20.csv every day. The file has these columns:
order_id,customer_id,order_date,amount
1001,C101,2026-10-20,49.95
1002,C102,2026-10-20,125.00
1003,C103,2026-10-20,-5.00
Our DAG will perform three tasks:
- Check and read the file for the Airflow logical date.
- Reject rows with missing IDs or a non-positive amount.
- Upsert the valid rows into a Postgres table.
The negative amount is intentional. It gives us something useful to validate rather than only moving rows from one place to another.
Prerequisites
You need Airflow running with the Postgres provider installed:
1
pip install apache-airflow-providers-postgres
Create a Postgres connection in Airflow with the connection ID orders_postgres. For a local demo, adding it under Admin -> Connections is fine. In production, I would create it through a secrets backend or deployment configuration.
The worker also needs access to /opt/airflow/data. If Airflow runs in Docker, mount the directory into every container that can execute the task. It is quite common for a file to exist on the scheduler but not on the worker.
Create the DAG
Create dags/daily_orders_etl.py:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
from datetime import datetime, timedelta
from pathlib import Path
from airflow.decorators import dag, task
from airflow.exceptions import AirflowFailException
from airflow.providers.postgres.hooks.postgres import PostgresHook
@dag(
dag_id="daily_orders_etl",
start_date=datetime(2026, 10, 1),
schedule="0 6 * * *",
catchup=False,
default_args={
"owner": "data-platform",
"retries": 2,
"retry_delay": timedelta(minutes=5),
},
tags=["orders", "etl"],
)
def daily_orders_etl():
pass
daily_orders_etl()
catchup=False means Airflow will not create every historical run between the start date and today when we enable the DAG. If we need a backfill, I prefer to request it deliberately rather than getting it by surprise.
Extract the Daily File
Add the extract task inside daily_orders_etl:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
@task
def extract(ds=None):
import csv
file_path = Path(f"/opt/airflow/data/orders_{ds}.csv")
if not file_path.exists():
raise AirflowFailException(f"Input file not found: {file_path}")
with file_path.open(newline="", encoding="utf-8") as file:
rows = list(csv.DictReader(file))
if not rows:
raise AirflowFailException("Input file is empty")
return rows
Airflow supplies ds from the run’s logical date. This is better than calling datetime.now() because a rerun for October 20 should still process the October 20 file. Wall-clock time works during the first test but becomes confusing during a backfill.
The task returns the rows, so Airflow puts them in XCom. This is acceptable for our tiny file. Do not return a million rows this way. For a larger file, write it to S3, GCS, or shared storage and return only its path.
Validate and Transform
Now add a task that converts values into the expected types and removes invalid records:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
@task
def transform(rows):
valid_rows = []
rejected = 0
for row in rows:
try:
amount = float(row["amount"])
if not row["order_id"] or not row["customer_id"]:
raise ValueError("missing identifier")
if amount <= 0:
raise ValueError("amount must be positive")
valid_rows.append({
"order_id": int(row["order_id"]),
"customer_id": row["customer_id"].strip(),
"order_date": row["order_date"],
"amount": amount,
})
except (KeyError, TypeError, ValueError):
rejected += 1
if not valid_rows:
raise AirflowFailException("No valid rows found")
print(f"Valid rows: {len(valid_rows)}, rejected rows: {rejected}")
return valid_rows
For our example, logging the rejected count is enough. In production, I would write rejected rows to a quarantine location with the validation reason. Otherwise, someone will ask why the source had 1,000 rows but the table received only 997, and a count alone will not help.
Load the Rows into Postgres
The load task creates the table and uses an upsert so rerunning the same Airflow run does not create duplicates:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
@task
def load(rows):
hook = PostgresHook(postgres_conn_id="orders_postgres")
hook.run("""
CREATE TABLE IF NOT EXISTS daily_orders (
order_id BIGINT PRIMARY KEY,
customer_id TEXT NOT NULL,
order_date DATE NOT NULL,
amount NUMERIC(12, 2) NOT NULL,
loaded_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
)
""")
values = [
(r["order_id"], r["customer_id"], r["order_date"], r["amount"])
for r in rows
]
sql = """
INSERT INTO daily_orders
(order_id, customer_id, order_date, amount)
VALUES (%s, %s, %s, %s)
ON CONFLICT (order_id) DO UPDATE SET
customer_id = EXCLUDED.customer_id,
order_date = EXCLUDED.order_date,
amount = EXCLUDED.amount,
loaded_at = NOW()
"""
connection = hook.get_conn()
with connection.cursor() as cursor:
cursor.executemany(sql, values)
connection.commit()
The load is idempotent because order_id is the primary key and a repeated order is updated. Airflow retries tasks, and operators also rerun failed tasks manually, so a plain append can create duplicates very easily.
Connect the Tasks
At the end of the DAG function, call the tasks like normal Python functions:
1
2
3
extracted_rows = extract()
transformed_rows = transform(extracted_rows)
load(transformed_rows)
The returned values are not the actual rows while Airflow parses the DAG. They are references to task outputs. Airflow uses those references to create dependencies and resolves the values only when the tasks run.
| Setting | Demo choice | Production consideration |
|---|---|---|
| Data between tasks | Rows in XCom | Store data externally and pass a URI |
| Credentials | Airflow connection | Secrets backend with rotation |
| Invalid rows | Log a count | Quarantine rows and alert on thresholds |
| Load behaviour | Upsert by order ID | Stage, validate, then merge in a transaction |
| File readiness | Fail if missing | Use a sensor with a sensible timeout |
Test Before Enabling the Schedule
First check that Airflow can parse the DAG:
1
2
airflow dags list-import-errors
airflow dags show daily_orders_etl
Then test each task for a fixed logical date:
1
2
3
airflow tasks test daily_orders_etl extract 2026-10-20
airflow tasks test daily_orders_etl transform 2026-10-20
airflow tasks test daily_orders_etl load 2026-10-20
After a successful run, verify the target rather than relying only on green boxes in the Airflow UI:
1
2
3
4
SELECT order_date, COUNT(*) AS row_count, SUM(amount) AS total_amount
FROM daily_orders
GROUP BY order_date
ORDER BY order_date DESC;
For a real pipeline, I would add a row-count check, alerting, and a task timeout. I would not keep table creation inside the daily DAG because schema changes should normally go through a migration process. It is included here only to make the first run easy to reproduce.
Conclusion
We created a small Airflow ETL DAG using the TaskFlow API, used the logical date to choose the input, validated the source rows, and made the Postgres load safe to retry. Once this works locally, the next step is to move larger intermediate data out of XCom, add proper data-quality checks, and deploy the connection and DAG through the same release process as the rest of the data platform.