Homework 05 — Scheduling, pools, and the two clocks

Lesson 05 · Data Engineering with Python · 3 tasks

These tasks extend the scheduled pipeline from the lab. You will add a daily-rollup DAG, constrain backfill concurrency with a Pool, and instrument a task to observe the difference between the logical date and physical execution time. All tasks assume the Airflow stack is still running from the lab session.

Task 1 — Add a daily-rollup DAG

Create a new file data-engineering/sample-pipeline/airflow/dags/gharchive_daily_rollup.py with the following properties:

schedule="@daily" and start_date=datetime(2026, 5, 1)
catchup=True, max_active_runs=4
A single task rollup that uses context["data_interval_start"] to compute the previous day's date string (e.g., "2026-05-01"), globs all 24 hourly output CSVs for that day from data/out/, reads them with pandas, groups by event type, and writes a daily summary Parquet to data/daily_rollups/<date>.parquet

Trigger a backfill for 2026-05-01 through 2026-05-03 and verify each day produces one Parquet file in data/daily_rollups/.

✓ Self-check: Grid view for gharchive_daily_rollup shows one cell per day; data/daily_rollups/2026-05-01.parquet exists and is readable by pd.read_parquet.

Task 2 — Constrain backfill concurrency with a Pool

Create an Airflow Pool named gharchive_api with 4 slots. Do this in the UI: Admin → Pools → + (add a new pool).

Then edit gharchive_hourly.py to wire the extract task to that pool:

extract = PythonOperator(
    task_id="extract",
    python_callable=_extract,
    pool="gharchive_api",
)

Trigger a backfill of 20 hours. Observe in Grid view that at most 4 extract tasks run concurrently, even if max_active_runs would allow more.

✓ Self-check: with max_active_runs=2 and the pool at 4 slots, you should see at most 2 extract tasks running at once (the more restrictive of the two limits wins). Raise max_active_runs to 8 temporarily, re-trigger a backfill of 20 hours, and now you should see at most 4 extract tasks — the pool is the binding constraint.

Task 3 — Observe the two clocks

Add the following print statement to the start of the _extract function:

def _extract(**context) -> None:
    from datetime import datetime as _dt
    from extract.gharchive import fetch_hour
    print(
        f"physical_now={_dt.utcnow()}, "
        f"data_interval_start={context['data_interval_start']}"
    )
    hour = _hour_key(context)
    fetch_hour(hour=hour)

Trigger a backfill of a single hour from yesterday. Open the Task Logs for the extract task of that run.

✓ Self-check: data_interval_start shows yesterday's date (or whichever past hour you backfilled); physical_now shows today's date/time. You can articulate the difference in one sentence: the data interval start is the start of the time window this run processes; the physical execution time is when the task actually ran — they differ for any historical replay.

View solutions →