Airflow Tutorial #1: Your First Airflow Pipeline

Goal: Build your first scheduled Airflow pipeline. It calls a public API for USD exchange rates, reshapes the data, and writes a dated file — automatically, every day, on a schedule you control.

Built on: Apache Airflow 3.3 via the Astro CLI. Everything runs locally in Docker. You do not install Airflow or Python directly.

Time: ~45–60 min (most of it one-time setup).

What you’ll have when you’re done

• A local Airflow 3.3 environment running on your machine.
• A DAG named exchange_rates_pipeline visible and runnable in the Airflow UI.
• A three-step pipeline (extract → transform → load) that produces a file like include/exchange_rates/rates_2026-07-17.json.
• The ability to read task logs and confirm your work in the UI.

Concepts cheatsheet:

Here are things to know. Click on links for more info for each.

• DAG (Directed Acyclic Graph) — the container for one workflow. “Directed” = tasks run in an order; “Acyclic” = no loops. Every pipeline is a DAG. → DAGs
• Task — one unit of work in a DAG (e.g. “extract”). → Tasks
• TaskFlow API (@dag / @task) — the modern way to write Airflow: decorate plain Python functions, and Airflow turns them into tasks and passes their return values between them. This is how you’ll write almost everything. → TaskFlow tutorial
• airflow.sdk — the stable import path for authoring DAGs in Airflow 3. Import dag, task, etc. from here. → Task SDK
• schedule — how often the DAG runs ("@daily", a cron string, None for manual-only). → Scheduling
• start_date — the first date the schedule is valid from. → DAG runs
• catchup — if True, on first unpause Airflow runs every missed interval since start_date. Set False to avoid an accidental flood of historical runs. → Catchup
• DAG run & task states — a “DAG run” is one execution of the whole DAG; each task instance has a state (success, failed, running, queued, up_for_retry). → DAG runs
• Logical date — the date a run represents (its data interval), not the wall-clock time it executed. Central to reproducibility; you’ll use it in the stretch goal. → What logical date means

Prerequisites (install these first)

You need four things. If you already have any, skip it.

1. A code editor. VS Code is recommended.
2. A terminal. macOS: Terminal. Windows: PowerShell. Linux: your shell.
3. A container runtime — Docker Desktop. Install it, launch it, and leave it running. Airflow runs inside Docker containers; if Docker isn’t running, astro dev start fails.
   • (macOS note: the Astro CLI can install Podman instead of Docker. Either works. If you have Docker Desktop running, the CLI uses it.)
4. The Astro CLI. Install per your OS below, then verify.

macOS (requires Homebrew):

brew install astro

Windows (PowerShell as Administrator):

winget install -e --id Astronomer.Astro

Linux:

curl -sSL install.astronomer.io | sudo bash

Verify (all OSes):

astro version

You should see a version number (e.g. 1.43.x). If the command isn’t found, close and reopen your terminal.

Install reference: Install the Astro CLI.

Step 1 — Create the project

Make a folder and initialize an Astro project inside it.

mkdir airflow-course
cd airflow-course
astro dev init

astro dev init scaffolds a standard project. The folders that matter for this lab:

• dags/ — where your DAG files live. You’ll add one here.
• include/ — supporting files. This folder is mounted into the container, so anything the pipeline writes here appears on your laptop. Your output lands here.
• Dockerfile — pins the Astro Runtime (and therefore the Airflow) version.
• requirements.txt — Python dependencies (unused in this lab; used in Lab 2).

→ Astro project structure

Open the airflow-course folder in VS Code now.

Step 2 — Start Airflow

From the airflow-course folder:

astro dev start

First run takes a few minutes (it downloads the Airflow image and starts containers for the scheduler, API server, DAG processor, and triggerer). When it finishes, it prints a local URL.

Open http://localhost:8080 in your browser. Log in:

• Username: admin
• Password: admin

You’re now looking at the Airflow 3 UI. → Airflow UI

(Useful commands: astro dev stop to pause, astro dev restart to rebuild after dependency changes, astro dev ps to see running containers.)

Step 3 — Create the DAG file

In the dags/ folder, create a new file named exactly:

exchange_rates_pipeline.py

Paste in the following. Read the comments — they explain each piece. Bonus points if you code it manually by hand first.

"""
Lab 1 — Hello, Pipeline
Daily USD exchange-rate snapshot: extract from API -> transform -> load to file.
"""
from __future__ import annotations

import json
import urllib.request
from datetime import datetime
from pathlib import Path

import pendulum

from airflow.sdk import dag, task

# This path is INSIDE the container. It is mounted to your local `include/`
# folder, so files written here show up on your laptop.
OUTPUT_DIR = Path("/usr/local/airflow/include/exchange_rates")

# Free, no-API-key exchange-rate endpoint. Returns JSON.
API_URL = "https://api.frankfurter.app/latest?base=USD&symbols=EUR,GBP,JPY,CAD,AUD"


@dag(
    schedule="@daily",                                   # run once a day
    start_date=pendulum.datetime(2025, 1, 1, tz="UTC"),  # schedule valid from here
    catchup=False,                                       # don't backfill old runs
    tags=["lab-1", "course"],                            # for filtering in the UI
)
def exchange_rates_pipeline():

    @task
    def extract() -> dict:
        """Call the API and return the raw JSON payload."""
        with urllib.request.urlopen(API_URL, timeout=30) as response:
            payload = json.loads(response.read().decode())
        print(f"Fetched rates dated {payload['date']}")
        return payload  # returned value is passed automatically to the next task

    @task
    def transform(payload: dict) -> list[dict]:
        """Reshape the nested JSON into a flat list of rows."""
        rows = [
            {
                "base": payload["base"],
                "target": currency,
                "rate": rate,
                "rate_date": payload["date"],
            }
            for currency, rate in payload["rates"].items()
        ]
        print(f"Transformed {len(rows)} rows")
        return rows

    @task
    def load(rows: list[dict]) -> str:
        """Write the rows to a dated JSON file."""
        OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
        # Uses wall-clock date. This is intentionally naive — the stretch goal fixes it.
        file_date = datetime.now().strftime("%Y-%m-%d")
        out_path = OUTPUT_DIR / f"rates_{file_date}.json"
        out_path.write_text(json.dumps(rows, indent=2))
        print(f"Wrote {len(rows)} rows to {out_path}")
        return str(out_path)

    # Define the order: extract -> transform -> load.
    # Passing return values wires the dependencies automatically.
    load(transform(extract()))


# This line registers the DAG with Airflow. Don't forget it.
exchange_rates_pipeline()

Save the file. Within ~30 seconds the DAG processor picks it up.

If the DAG doesn’t appear or shows an error: run astro dev parse in your terminal — it reports syntax/import errors without you hunting through the UI.

Step 4 — Run it

1. Go to http://localhost:8080. On the DAGs list, find exchange_rates_pipeline.
2. Unpause it: click the toggle on the left so it’s on (blue).
3. Trigger it manually: click the DAG name to open it, then click the ▶ Trigger button (top right).
4. Watch the run appear and its tasks turn from light (queued/running) to dark green (success).

Step 5 — Check your work

In the UI:

• Grid view (default DAG tab) — each column is a run, each row is a task. Green = success, red = failed. Click any task square to inspect it.
• Logs — click the load task square → Logs tab. You should see Wrote 5 rows to /usr/local/airflow/include/exchange_rates/rates_YYYY-MM-DD.json. Logs are where you debug everything, always.
• Graph view — see extract → transform → load as connected boxes. This is your dependency chain.
• Code view — the source Airflow parsed, so you can confirm it’s running what you think.
• XCom (click a task → XCom tab) — the return values being passed between tasks. This is a preview of Lab 3.

On your laptop:

Open the include/exchange_rates/ folder in VS Code. There’s a rates_YYYY-MM-DD.json file with five rows (EUR, GBP, JPY, CAD, AUD). That’s your pipeline’s output.

Ship it (definition of done)

• exchange_rates_pipeline shows in the UI, unpaused.
• A manual run completes with all three tasks green.
• You found the Wrote 5 rows... line in the load task’s logs.
• The output JSON file exists in include/exchange_rates/.

Stretch Goal 1 — Use the logical date, not the wall clock

Why: datetime.now() stamps the file with when the code ran. But a run represents a date (its logical date), and runs can be re-run or backfilled. If you use wall-clock time, re-running yesterday’s pipeline writes today’s filename — you can’t reproduce anything. Fix it to use the run’s logical date.

What to change: the load task. Airflow passes context variables (including ds, the logical date as a YYYY-MM-DD string) into any task that accepts **context. → Context variables

Where: in dags/exchange_rates_pipeline.py, replace the load function with this version:

    @task
    def load(rows: list[dict], **context) -> str:
        """Write the rows to a file named for the run's LOGICAL date."""
        OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
        file_date = context["ds"]  # logical date 'YYYY-MM-DD' — reproducible on re-run
        out_path = OUTPUT_DIR / f"rates_{file_date}.json"
        out_path.write_text(json.dumps(rows, indent=2))
        print(f"Wrote {len(rows)} rows to {out_path} (logical date {file_date})")
        return str(out_path)

How to run: save the file (it reloads automatically). In the UI, trigger the DAG again.

Where to see the impact:

• In the load logs, the filename now matches the run’s Logical Date — check it against the logical date shown in the run details / Grid tooltip.
• Now re-run the same run: in Grid view, click the load task square → Clear task (confirm). It re-executes and writes the same filename, overwriting the same file. That’s idempotency — the property that re-running produces the same result, not a duplicate. With the old datetime.now() version, a re-run on a different day would have created a different file. That difference is the whole point.

Stretch Goal 2 — Add a data-quality check

Why: never load empty or broken data silently. Add a task that fails the run if there are zero rows.

What to change: add a new validate task and wire it between transform and load.

Where: add this task inside the DAG (e.g. right after transform):

    @task
    def validate(rows: list[dict]) -> list[dict]:
        """Fail the run if there's no data."""
        if len(rows) == 0:
            raise ValueError("No rows returned from API — failing the run.")
        print(f"Validation passed: {len(rows)} rows")
        return rows

Then change the final wiring line from:

    load(transform(extract()))

to:

    load(validate(transform(extract())))

How to run: save, then trigger the DAG. All four tasks should go green (extract → transform → validate → load), and Graph view now shows the new validate box in the chain.

Where to see the impact — make it fail on purpose: temporarily change the check to if len(rows) < 100: (you only get 5 rows, so this trips). Save, trigger, and watch:

• In Grid view, the validate task turns red, and load shows as upstream_failed (it never runs — a failed upstream stops the chain).
• Click the red validate square → Logs → you’ll see the ValueError traceback. This is exactly how you’ll diagnose real failures.

Then revert the check to if len(rows) == 0:, save, and trigger once more to confirm it’s green again.

The trap (don’t skip this)

Never use datetime.now() (or pendulum.now()) to stamp or partition data inside a task. Airflow runs are keyed to a logical date, and runs get re-executed and backfilled constantly in real life. Wall-clock time makes your pipeline non-reproducible: the same logical run produces different output depending on when it happened to execute. Always pull the date from the run context (ds, logical_date, data_interval_start). You built the habit in Stretch Goal 1 — keep it for every lab after this.

Shut down

When you’re done:

astro dev stop

This stops the containers but keeps your project and metadata. Run astro dev start again anytime to pick up where you left off.

What’s next — Lab 2 preview

Right now your data lands in a file. Next lab, “The Reliable Loader,” you’ll load it into a Postgres database using Connections, Hooks, and a provider package — and make the whole thing idempotent so re-running never creates duplicates.