First Airflow shift

Do this after Your first DAG and Operators. Keep the stack from Getting Started running.

You will orchestrate a tiny retail extract: read a CSV, total amount by store, write a summary file. The website runs nothing. Data stays a handful of rows — XCom only passes the file path.

What you need

• Airflow UI at http://127.0.0.1:18080 (user airflow / airflow)
• ~/dcubes/airflow-lab with dags/ and data/ bind-mounted

If the stack is down:

cd ~/dcubes/airflow-lab
docker compose up -d

1. Land the CSV

Download orders.csv or write it:

mkdir -p ~/dcubes/airflow-lab/data
cat > ~/dcubes/airflow-lab/data/orders.csv << 'EOF'
order_id,store,amount
1,north,12.50
2,south,8.00
3,north,20.00
4,east,5.25
5,south,11.75
EOF

Inside the container that path is /opt/airflow/data/orders.csv (see Getting Started volumes).

ls -l ~/dcubes/airflow-lab/data/orders.csv

2. Write the DAG

cat > ~/dcubes/airflow-lab/dags/retail_shift.py << 'EOF'
import csv
from collections import defaultdict
from datetime import datetime
from pathlib import Path

from airflow.sdk import dag, task

DATA = Path("/opt/airflow/data")


@dag(
    schedule=None,
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=["dcubes", "shift"],
)
def retail_shift():
    @task
    def extract() -> str:
        src = DATA / "orders.csv"
        if not src.is_file():
            raise FileNotFoundError(f"missing {src}")
        print(f"extract {src}")
        return str(src)

    @task
    def transform(path: str) -> str:
        totals: dict[str, float] = defaultdict(float)
        with Path(path).open(newline="") as handle:
            for row in csv.DictReader(handle):
                totals[row["store"]] += float(row["amount"])
        out = DATA / "store_totals.csv"
        with out.open("w", newline="") as handle:
            writer = csv.writer(handle)
            writer.writerow(["store", "amount"])
            for store in sorted(totals):
                writer.writerow([store, f"{totals[store]:.2f}"])
        print(f"wrote {out}")
        return str(out)

    @task
    def load(path: str) -> None:
        text = Path(path).read_text()
        summary = DATA / "summary.txt"
        summary.write_text("store totals\n" + text)
        print(summary.read_text())

    load(transform(extract()))


retail_shift()
EOF

Wait 30–60 seconds. Refresh the UI. You want retail_shift.

If it is missing:

docker compose -f ~/dcubes/airflow-lab/compose.yaml logs airflow-dag-processor --tail 40

(or cd ~/dcubes/airflow-lab && docker compose logs airflow-dag-processor --tail 40)

3. Trigger and inspect

1. Unpause retail_shift.
2. Trigger.
3. Graph: extract → transform → load, all success.
4. extract log mentions /opt/airflow/data/orders.csv.
5. load log prints store totals.

On the host:

cat ~/dcubes/airflow-lab/data/store_totals.csv
cat ~/dcubes/airflow-lab/data/summary.txt

store,amount
east,5.25
north,32.50
south,19.75

north is 12.50 + 20.00. If a number is wrong, the DAG is not the database — fix the CSV or the transform and clear transform + downstream.

4. Optional: version the DAG

You already have Git.

cd ~/dcubes/airflow-lab
printf '%s\n' '.env' 'logs/' 'config/' 'data/' '__pycache__/' > .gitignore
git init
git add compose.yaml dags/hello_airflow.py dags/retail_shift.py .gitignore
git status
git commit -m "Add Airflow lab compose and retail_shift DAG"

Do not commit .env, logs/, or data/. Compose and DAG files are the source of truth.

Checklist

• retail_shift appears in the UI
• One successful run, three green tasks
• data/store_totals.csv matches the example
• XCom values are paths, not the CSV contents

Common mistakes

• CSV on the host but not under data/ — the container only sees the bind mount
• Returning the parsed rows from extract — return the path
• Leaving fail_then_fix.py as this is not python — delete it so the processor stays clean

Next

Airflow is the first Data Engineering module. The lakehouse lab is next: the same machine, a bigger Compose stack. Warehouses and modeling stay coming soon. Still guidance only, still on your machine.

When you are done with the lab for the day:

cd ~/dcubes/airflow-lab
docker compose down