Solutions — Homework 05

Lesson 05 · Data Engineering with Python

Task 1 — Daily-rollup DAG

The key is using data_interval_start (not datetime.utcnow()) to derive the day being rolled up. A daily DAG with start_date=2026-05-01 fires for the first time on 2026-05-02, and its data_interval_start is 2026-05-01T00:00:00 — that is the day it should process. This is the "fires after the interval ends" behaviour: the run for May 1 fires at the end of May 1 (i.e., the start of May 2).

from __future__ import annotations

import glob
import pathlib
import sys
from datetime import datetime

from airflow import DAG
from airflow.operators.python import PythonOperator

_SAMPLE_PIPELINE_ROOT = pathlib.Path("/opt/airflow/sample-pipeline")
if _SAMPLE_PIPELINE_ROOT.exists() and str(_SAMPLE_PIPELINE_ROOT) not in sys.path:
    sys.path.insert(0, str(_SAMPLE_PIPELINE_ROOT))


def _rollup(**context) -> None:
    import pandas as pd
    day = context["data_interval_start"].strftime("%Y-%m-%d")
    files = sorted(
        glob.glob(f"/opt/airflow/sample-pipeline/data/out/{day}-*.csv")
    )
    if not files:
        print(f"no output files for {day}, skipping")
        return
    df = pd.concat(pd.read_csv(f) for f in files)
    out = df.groupby("event_type").size().reset_index(name="count")
    out_dir = pathlib.Path("/opt/airflow/sample-pipeline/data/daily_rollups")
    out_dir.mkdir(parents=True, exist_ok=True)
    out.to_parquet(out_dir / f"{day}.parquet", index=False)
    print(f"wrote {len(out)} rows to daily_rollups/{day}.parquet")


with DAG(
    dag_id="gharchive_daily_rollup",
    start_date=datetime(2026, 5, 1),
    schedule="@daily",
    catchup=True,
    max_active_runs=4,
    tags=["data-engineering", "module-2"],
    default_args={"owner": "data-eng"},
) as dag:
    PythonOperator(task_id="rollup", python_callable=_rollup)

Task 2 — Constraining concurrency with a Pool

Create the pool in the UI: Admin → Pools → +. Set name to gharchive_api and slots to 4. Then add pool="gharchive_api" to the PythonOperator:

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

With max_active_runs=2 and the pool at 4 slots, the binding constraint is max_active_runs: at most 2 DagRuns are active, each with 1 extract task, so at most 2 extract tasks run. Raise max_active_runs to 8 and the pool becomes the binding limit: 4 extract tasks run concurrently regardless of how many DagRuns are active.

Task 3 — Observing the two clocks

After adding the print statement and triggering a backfill of a past hour, your Task Logs should show something like:

physical_now=2026-05-29 09:15:42.123456, data_interval_start=2026-05-01 00:00:00+00:00

The one-sentence explanation: data_interval_start is the start of the time window this run is processing — it is fixed for the DagRun and does not change on retry or replay; datetime.utcnow() is when the task actually executed — it changes every time the task runs.

Lesson 06 fixes the silent bug exposed in the lab's Step 7: the same interval re-run twice produces non-deterministic output unless the load is transactional. We address that next.