Homework 09 — Solutions

Lesson 09 · Data Engineering with Python · Worked solutions

Solution to Task 1 — weekly_repo_growth view

Add the following view definition inside open_pipeline_db() in query.py, immediately after the events view:

CREATE OR REPLACE VIEW weekly_repo_growth AS
WITH push_per_week AS (
  SELECT repo.name AS repo,
         date_trunc('week', date::date) AS iso_week,
         count(*) AS pushes
  FROM events
  WHERE type = 'PushEvent'
  GROUP BY repo, iso_week
)
SELECT repo,
       iso_week,
       pushes,
       pushes - lag(pushes) OVER (PARTITION BY repo ORDER BY iso_week) AS delta
FROM push_per_week;

Solution to Task 2 — top_repos.py with --out

The relevant diff to top_repos.py (the changed sections only):

def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--since", required=True, help="YYYY-MM-DD inclusive")
    parser.add_argument("--until", required=True, help="YYYY-MM-DD inclusive")
    parser.add_argument("--n", type=int, default=10)
    parser.add_argument("--out", default=None, help="Write result to this Parquet path")
    args = parser.parse_args()

    with open_pipeline_db() as con:
        if args.out:
            con.execute(
                f"""
                COPY (
                  SELECT repo.name AS repo, count(*) AS n
                  FROM events
                  WHERE date BETWEEN ? AND ?
                  GROUP BY repo
                  ORDER BY n DESC
                  LIMIT ?
                ) TO '{args.out}' (FORMAT PARQUET);
                """,
                [args.since, args.until, args.n],
            )
        else:
            rows = con.execute(
                """
                SELECT repo.name AS repo, count(*) AS n
                FROM events
                WHERE date BETWEEN ? AND ?
                GROUP BY repo
                ORDER BY n DESC
                LIMIT ?
                """,
                [args.since, args.until, args.n],
            ).fetchall()
            for repo, count in rows:
                print(f"{repo}\t{count}")

Solution to Task 3 — pandas vs DuckDB benchmark

"""bench_pandas_vs_duckdb.py — Compare count-by-type via pandas vs DuckDB."""
import time
import duckdb
import pandas as pd
import pathlib

PARQUET_ROOT = pathlib.Path("data-engineering/sample-pipeline/data/parquet")


def pandas_path():
    files = sorted(PARQUET_ROOT.glob("date=*/*.parquet"))
    df = pd.concat([pd.read_parquet(p) for p in files])
    return df.groupby("type").size().to_dict()


def duckdb_path():
    con = duckdb.connect(":memory:")
    return dict(
        con.execute(
            f"SELECT type, count(*) FROM read_parquet('{PARQUET_ROOT}/date=*/*.parquet',"
            " hive_partitioning=true) GROUP BY type"
        ).fetchall()
    )


def bench(label, fn, runs=3):
    times = []
    for _ in range(runs):
        t0 = time.perf_counter()
        fn()
        times.append(time.perf_counter() - t0)
    best = min(times)
    print(f"  {label}: best of {runs} = {best:.3f}s")
    return best


print("Benchmarking...")
p = bench("pandas", pandas_path)
d = bench("duckdb", duckdb_path)
print(f"\npandas={p:.3f}s  duckdb={d:.3f}s  ratio={p/d:.1f}x")
# COMMENT: DuckDB wins because pandas reads every Parquet column into RAM
# before grouping; DuckDB applies column projection (reads only 'type') and
# vectorised aggregation, so far less data is decompressed.