From a Python script to a scheduled, monitored, replayable pipeline.
Data Engineering Module 2 OrchestrationA cron line is fine until it isn't. Here's what breaks first:
Schedule · Dependency · Retry · Observability
Apache Airflow is an open-source workflow scheduler originally created at Airbnb in 2014 and donated to the Apache Software Foundation in 2016, where it graduated as a top-level project. It is now the dominant data-pipeline orchestrator in the industry.
| Abstraction | What it is | Example |
|---|---|---|
| DAG | The workflow — a Python file declaring tasks and their execution order | gharchive_hourly |
| Task | One unit of work; one task instance is created per DagRun | extract, transform |
| Operator | The type of task — defines what code runs when the task executes | PythonOperator, BashOperator, BranchPythonOperator |
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
with DAG(
dag_id="gharchive_hourly",
start_date=datetime(2026, 5, 1),
schedule=None, # we'll schedule it in Lesson 05
catchup=False,
) as dag:
extract = PythonOperator(task_id="extract", python_callable=extract_hour)
transform = PythonOperator(task_id="transform", python_callable=transform_hour)
extract >> transform # task dependency>> is Airflow's "set downstream" operator. extract >> transform means: run transform only after extract succeeds.
All four components run as Docker containers on your laptop via docker compose up. The LocalExecutor is sufficient for development workloads.
cd data-engineering/sample-pipeline/airflow
docker compose up -d
open http://localhost:8080 # admin / adminapache/airflow:2.9.3-python3.11 imageairflow db migrateAirflow continuously scans the dags/ directory every 30 seconds (configurable via dag_dir_list_interval). Each .py file that contains a DAG object becomes a discoverable workflow in the UI.
.py file into dags/ — it appears in the UI within ~30svolumes: block in docker-compose.yaml mounts ./dags into the container so edits on your host are immediately visible inside AirflowThe first DAG we add is gharchive_hourly — a thin wrapper that calls the Module 1 extract_hour and transform_hour functions.
gharchive_hourly → click the "Play" button → "Trigger DAG"airflow dags trigger gharchive_hourly (inside the container or via docker compose exec)In Lesson 04 the DAG has schedule=None — it only ever runs when you explicitly trigger it. This is intentional: learn the mechanics before adding a schedule.
none → scheduled → queued → running → success
On failure: failed → (if retries > 0) up_for_retry → scheduled → queued → running → ...
Or: skipped (upstream failed or branched away)
Every box in the UI Grid view represents one task instance in one of these states. Click any box to see the full lifecycle history and logs for that attempt.
The DAG is a thin wrapper — it does not duplicate the extract or transform logic. It calls the functions that already exist in sample-pipeline/extract/ and sample-pipeline/transform/.
def _extract(**context) -> None:
"""L04: hardcoded hour. L05 replaces with context['data_interval_start']."""
from extract.gharchive import fetch_hour
fetch_hour(hour="2026-05-01-12") # hardcoded for L04; parameterized in L05Note the deferred import inside the function body. Airflow re-imports every DAG file every 30s — keeping heavy imports inside the callable keeps the scheduler heartbeat fast.
context["params"]volumes: block in docker-compose.yaml is wrong, the UI shows zero DAGs; check the "DAG Import Errors" banner for cluesstart_date in the future — the scheduler refuses to create runs for a DAG whose start_date is still in the future; always use a past fixed datecron alone cannot