Lesson 04 Airflow Basics
Theme:

Solutions — Homework 04

Lesson 04 · Data Engineering with Python

Task 1 — Add a third task: summarize

Add the following callable and operator to gharchive_hourly.py. The import for pandas is deferred inside the function body — not at module scope — to keep parse time fast. 

def _summarize(**context) -> None:
    import pandas as pd
    df = pd.read_parquet("/opt/airflow/sample-pipeline/data/2026-05-01-12.parquet")
    print(df.groupby("event_type").size().sort_values(ascending=False).head(5).to_string())

# inside the DAG context (after the existing operators):
summarize = PythonOperator(task_id="summarize", python_callable=_summarize)
extract >> transform >> summarize

Common mistake

Importing pandas at module scope (import pandas as pd at the top of the file) works, but adds ~100 ms to every scheduler heartbeat. Over a deployment with dozens of DAGs, this adds up. Keep heavy imports inside the function body.

Task 2 — Bash operator practice: cleanup

Import BashOperator and add the fourth task. Note: the import can go at the top of the file alongside the existing Airflow imports — BashOperator itself is lightweight to import. 

from airflow.operators.bash import BashOperator

# inside the DAG context:
cleanup = BashOperator(
    task_id="cleanup",
    bash_command="du -sh /opt/airflow/sample-pipeline/data/",
)
extract >> transform >> summarize >> cleanup

Common mistake

Using a host path (e.g., data-engineering/sample-pipeline/data/) instead of the container path. BashOperator runs inside the Airflow worker container — paths must reference container mount points. The docker-compose.yaml mounts the sample-pipeline/ directory at /opt/airflow/sample-pipeline, so that is the correct base path.

Task 3 — Read the parse log

A sample line from the scheduler logs: 

[2026-05-26 09:10:01] {dag_processing.py:712} INFO - Processing dags/gharchive_hourly.py took 0.34 seconds

Your number will vary. On a clean laptop with deferred imports, expect 0.1–0.5 seconds. If you see > 1 second, check whether you have any heavy imports at module scope.

Common mistake

Confusing parse time with task runtime. Parse time is the time the scheduler spends importing the DAG file every 30 seconds — it is purely the cost of Python import statements. Task runtime is how long the actual callable takes to execute (downloading a file, running a transformation). These are completely separate metrics and appear in different log locations: parse time in airflow-scheduler logs, task runtime in the task instance logs accessible from the Grid view.

Lesson 05 makes this DAG scheduled and replayable across past time windows. Instead of a hardcoded hour, each run will process its own data interval — making the pipeline genuinely useful and backfillable.