Lab 09 — Query the pipeline’s Parquet partitions with DuckDB

Lesson 09 · Data Engineering with Python · ~45 min

Goal: by the end of this lab you will have (a) installed DuckDB from the updated requirements.txt, (b) run count_by_type.py and top_repos.py against the L02 Parquet partitions, (c) extended query.py with a second view, and (d) wrapped a DuckDB query in an Airflow PythonOperator as a sandbox exercise.

Prerequisites

• L02 Parquet partitions exist at data-engineering/sample-pipeline/data/parquet/date=*/.
• The Module 1–3 requirements.txt is installed in your virtualenv. This lab’s first step picks up the Module 4 duckdb>=0.10 line.
• One terminal from which you can run python -m … commands.

Step 1 — Install DuckDB and confirm the import

From the repo root, run:

pip install -r data-engineering/sample-pipeline/requirements.txt

If you only want the Module 4 dependency without reinstalling everything:

pip install 'duckdb>=0.10'

Confirm the install:

python -c "import duckdb; print(duckdb.__version__)"

Expected: a version string of 0.10 or higher.

Step 2 — Walk through query.py

Open data-engineering/sample-pipeline/duckdb/query.py in your editor. Read it line by line:

• PARQUET_ROOT is resolved relative to the file’s location on disk, not the current working directory. The same script runs correctly whether you invoke it from the repo root, from inside sample-pipeline/, or from anywhere else.
• connect(":memory:") — no persistent file. The view exists for the lifetime of the connection object only. When the connection closes (end of with block), all state is gone. This is intentional: each pipeline run gets a fresh, deterministic slate.
• hive_partitioning = true — the critical flag. It tells DuckDB to parse the date=YYYY-MM-DD/ directory names and expose date as a column, enabling partition pruning.
• CREATE OR REPLACE VIEW vs CREATE TABLE AS SELECT — a view is cheap to create; the scan is deferred until you actually query it. A CREATE TABLE would copy all the Parquet data into DuckDB’s in-memory store and defeat the “read in place” goal.

Step 3 — Run the two query scripts

cd data-engineering/sample-pipeline
python -m duckdb.count_by_type

Expected: a sorted list of GitHub event types with counts. PushEvent will dominate on most days.

python -m duckdb.top_repos --since 2024-01-01 --until 2024-01-07 --n 10

Expected: up to 10 repo TAB n rows for the highest-volume repositories in the window.

Step 4 — Add a daily_counts view

Edit query.py and add a second CREATE OR REPLACE VIEW immediately after the events view:

con.execute("""
    CREATE OR REPLACE VIEW daily_counts AS
    SELECT date, type, count(*) AS n
    FROM events
    GROUP BY date, type;
""")

Verify it in a one-off Python script:

import sys
sys.path.insert(0, "data-engineering/sample-pipeline")
from duckdb.query import open_pipeline_db

with open_pipeline_db() as con:
    print(con.execute(
        "SELECT * FROM daily_counts WHERE type = 'PullRequestEvent' "
        "ORDER BY date DESC LIMIT 5"
    ).fetchall())

Expected: up to five rows of (date, type, n) tuples.

Step 5 — Wrap it in an Airflow PythonOperator

This step is a sandbox exercise. Add a new task to the Module 2 DAG file sample-pipeline/airflow/dags/etl_dag.py:

from airflow.operators.python import PythonOperator
from duckdb.count_by_type import main as count_by_type

count_task = PythonOperator(
    task_id="duckdb_count_by_type",
    python_callable=count_by_type,
    dag=dag,
)
# Wire after the existing load-partitions task:
# load_task >> count_task

Trigger a manual DAG run and confirm the task logs the same output as the CLI invocation.

Step 6 — Spot-check predicate pushdown

Enable DuckDB’s profiling output to confirm that partition pruning is actually happening:

import sys
sys.path.insert(0, "data-engineering/sample-pipeline")
from duckdb.query import open_pipeline_db

with open_pipeline_db() as con:
    con.execute("PRAGMA enable_profiling = 'text'")
    con.execute("PRAGMA profile_output = '/tmp/duckdb_profile.txt'")
    con.execute(
        "SELECT count(*) FROM events WHERE date = '2024-01-01'"
    ).fetchall()

Open /tmp/duckdb_profile.txt. Locate the READ_PARQUET node and confirm it lists only the matching partition’s file(s) under “Files Opened” — not every partition in the directory.

What you should leave with

Troubleshooting