Lab 02 — Convert the L01 output to partitioned Parquet and measure

Lesson 02 · Data Engineering with Python · ~50 min

Goal: take the L01 pipeline's raw output, rewrite it as partitioned Parquet, and measure the differences in file size and query latency between formats.

What you need

• The L01 virtual environment with pyarrow already installed (it's in requirements.txt — run pip install -r requirements.txt from data-engineering/sample-pipeline/ if you skipped it)
• At least one hour of gharchive data already downloaded to data/raw/ (for example, data/raw/2024-01-01-15.json.gz)
• ~200 MB of free disk space for the Parquet output

✓ Self-check: python -c "import pyarrow.parquet as pq; print(pq.__version__)" prints a version without errors.

Step 1 — Add to_parquet to transform/events.py

Open data-engineering/sample-pipeline/transform/events.py and add the following function below your existing count_by_type and write_csv functions:

import pyarrow as pa
import pyarrow.parquet as pq

def to_parquet(json_gz: pathlib.Path, parquet_out: pathlib.Path) -> None:
    """Parse a gharchive .json.gz and write it as a Parquet file.

    Keeps four columns: type, actor, repo, created_at.
    """
    rows = []
    with gzip.open(json_gz, "rt") as f:
        for line in f:
            ev = json.loads(line)
            rows.append({
                "type":       ev["type"],
                "actor":      ev["actor"]["login"],
                "repo":       ev["repo"]["name"],
                "created_at": ev["created_at"],
            })
    table = pa.Table.from_pylist(rows)
    pq.write_table(table, parquet_out, compression="snappy")

Also add an argparse flag so the script accepts a --format option:

if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("--input",  required=True, type=pathlib.Path)
    parser.add_argument("--output", required=True, type=pathlib.Path)
    parser.add_argument("--format", choices=["csv", "parquet"], default="csv")
    args = parser.parse_args()

    args.output.parent.mkdir(parents=True, exist_ok=True)

    if args.format == "csv":
        write_csv(count_by_type(args.input), args.output)
    else:
        to_parquet(args.input, args.output)

Step 2 — Run for one hour

python -m transform.events \
    --input  data/raw/2024-01-01-15.json.gz \
    --output data/parquet/2024-01-01-15.parquet \
    --format parquet
ls -lh data/parquet/2024-01-01-15.parquet

✓ Expected: the file is between 1 MB and 3 MB. Compare it to the matching .csv file from L01 — the Parquet file is smaller despite containing four columns instead of one.

Step 3 — Run for 24 hours, partitioned by date

Now modify to_parquet to use pq.write_to_dataset, which handles partitioned layouts automatically:

def to_parquet_partitioned(
    json_gz: pathlib.Path,
    dataset_root: pathlib.Path,
    partition_date: str,          # e.g. "2024-01-01"
) -> None:
    rows = []
    with gzip.open(json_gz, "rt") as f:
        for line in f:
            ev = json.loads(line)
            rows.append({
                "type":       ev["type"],
                "actor":      ev["actor"]["login"],
                "repo":       ev["repo"]["name"],
                "created_at": ev["created_at"],
                "date":       partition_date,   # the partition column
            })
    table = pa.Table.from_pylist(rows)
    pq.write_to_dataset(
        table,
        root_path=str(dataset_root),
        partition_cols=["date"],
        compression="snappy",
    )

Run it for all 24 hours of 2024-01-01:

from pathlib import Path
for hour in range(24):
    src = Path(f"data/raw/2024-01-01-{hour}.json.gz")
    if src.exists():
        to_parquet_partitioned(src, Path("data/parquet"), "2024-01-01")

Expected output layout:

data/parquet/
  date=2024-01-01/
    part-0.parquet

✓ Self-check: ls data/parquet/date=2024-01-01/ shows at least one .parquet file.

Step 4 — Measure read latency

Open a Python REPL inside the sample-pipeline directory:

import time, pandas as pd, duckdb

# --- pandas ----------------------------------------------------------
t = time.perf_counter()
df = pd.read_parquet("data/parquet/date=2024-01-01/")
elapsed = round(time.perf_counter() - t, 3)
print(f"pandas read: {elapsed} s, {len(df):,} rows")

# --- DuckDB ----------------------------------------------------------
t = time.perf_counter()
n = duckdb.query(
    "SELECT COUNT(*) FROM 'data/parquet/date=2024-01-01/*.parquet' WHERE type='PushEvent'"
).fetchone()[0]
elapsed = round(time.perf_counter() - t, 3)
print(f"duckdb count: {elapsed} s, {n:,} matching rows")

✓ Expected: pandas takes a few seconds (it reads the whole dataset into memory). DuckDB's count is sub-second — often under 0.1 s — thanks to predicate pushdown on the type column statistics.

Step 5 — Compare to the L01 CSV

du -h data/out/*.csv data/parquet/date=2024-01-01/

Notice how much smaller the Parquet partition is, despite containing four full columns (type, actor, repo, created_at) while the L01 CSV contained only a two-column aggregated summary (type count rollup with ~30 rows). The difference is compression and binary encoding — Parquet wins on both size and richness.

What you should leave with

• A data/parquet/date=YYYY-MM-DD/ partitioned dataset you can query without loading into memory
• A concrete sense of the latency gap between pandas (full read) and DuckDB (predicate pushdown)
• Muscle-memory for pa.Table.from_pylist, pq.write_table, and pq.write_to_dataset

Troubleshooting

schema = pa.schema([
    ("type",       pa.string()),
    ("actor",      pa.string()),
    ("repo",       pa.string()),
    ("created_at", pa.string()),
    ("date",       pa.string()),
])
table = pa.Table.from_pylist(rows, schema=schema)