Lab 01 — Implement the 50-line gharchive ETL

Lesson 01 · Data Engineering with Python · ~45 min

Goal: end this lab with a working command python -m extract.gharchive --hour 2024-01-01-15 && python -m transform.events --input data/raw/2024-01-01-15.json.gz --output data/out/2024-01-01-15.csv that downloads one hour of gharchive data and writes a CSV counting every event type.

What you need

• Python 3.10 or newer
• ~500 MB of free disk space (one .json.gz + CSV output)
• An internet connection (for the initial download)

Step 1 — Set up the venv

Open a terminal and navigate to the sample-pipeline directory:

cd data-engineering/sample-pipeline
python -m venv .venv
source .venv/bin/activate      # Windows: .venv\Scripts\activate
pip install -r requirements.txt

The install step pulls in requests, pandas, pyarrow, and other dependencies for the whole course. It takes a minute or two on the first run.

✓ Self-check: python -c "import requests, pandas, pyarrow" runs without errors.

Step 2 — Implement fetch_hour

Open extract/gharchive.py. Find the fetch_hour function — it currently raises NotImplementedError. Replace the body with the implementation from Slide 5:

import requests, pathlib

URL = "https://data.gharchive.org/{hour}.json.gz"

def fetch_hour(hour: str, out_dir: pathlib.Path) -> pathlib.Path:
    out_dir.mkdir(parents=True, exist_ok=True)
    dest = out_dir / f"{hour}.json.gz"
    if dest.exists():
        return dest                      # poor man's caching
    resp = requests.get(URL.format(hour=hour), timeout=30)
    resp.raise_for_status()
    dest.write_bytes(resp.content)
    return dest

Also update the module-level main() function to accept an --hour argument and call fetch_hour:

def main() -> None:
    import argparse
    parser = argparse.ArgumentParser(description="Download one gharchive hour")
    parser.add_argument("--hour", required=True, help='e.g. "2024-01-01-15"')
    args = parser.parse_args()
    path = fetch_hour(args.hour, pathlib.Path("data/raw"))
    print(f"downloaded: {path}")

Test from the terminal (still inside data-engineering/sample-pipeline):

python -m extract.gharchive --hour 2024-01-01-15
# expected: prints data/raw/2024-01-01-15.json.gz, file is roughly 50 MB
ls -lh data/raw/

✓ Self-check: data/raw/2024-01-01-15.json.gz exists and is between 20 MB and 200 MB.

Step 3 — Implement count_by_type

Open transform/events.py. Replace the NotImplementedError body of count_by_type with the implementation from Slide 6:

import gzip, json, pathlib
from collections import Counter

def count_by_type(path: pathlib.Path) -> Counter:
    counts = Counter()
    with gzip.open(path, "rt") as f:
        for line in f:
            ev = json.loads(line)
            counts[ev["type"]] += 1
    return counts

Try it in a Python REPL:

python -c "
from transform.events import count_by_type
import pathlib
result = count_by_type(pathlib.Path('data/raw/2024-01-01-15.json.gz'))
print(dict(list(result.items())[:5]))
print('total events:', sum(result.values()))
"

✓ Self-check: output is a dict like {'PushEvent': 45212, 'CreateEvent': 12103, ...} and the total is in the tens of thousands.

Step 4 — Add write_csv and wire __main__

Add the write_csv function (Slide 7) to transform/events.py:

import csv

def write_csv(counts: Counter, path: pathlib.Path) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("w", newline="") as f:
        w = csv.writer(f)
        w.writerow(["type", "count"])
        for evtype, n in counts.most_common():
            w.writerow([evtype, n])

Update the module-level main() to accept --input and --output arguments:

def main() -> None:
    import argparse
    parser = argparse.ArgumentParser(description="Transform gharchive events to CSV")
    parser.add_argument("--input", required=True, help="Path to .json.gz input file")
    parser.add_argument("--output", required=True, help="Path to .csv output file")
    args = parser.parse_args()
    counts = count_by_type(pathlib.Path(args.input))
    write_csv(counts, pathlib.Path(args.output))
    print(f"wrote {sum(counts.values())} events across {len(counts)} types")

Run the full pipeline:

python -m transform.events \
  --input  data/raw/2024-01-01-15.json.gz \
  --output data/out/2024-01-01-15.csv
head -5 data/out/2024-01-01-15.csv

✓ Self-check: the CSV starts with a type,count header and the first data row is PushEvent,<some large number>.

Step 5 — Re-run and notice the cache

Run the extract step a second time:

python -m extract.gharchive --hour 2024-01-01-15

It should return instantly — the file is already on disk, so fetch_hour skips the download and returns the cached path. This is your first taste of idempotency: running the same operation twice produces the same result without repeating expensive work.

✓ Self-check: the second run completes in under 1 second.

Step 6 — Try a different hour

Pick any hour from 2024 — but choose one that is at least three hours in the past (gharchive publishes with roughly a two-hour lag). For example:

python -m extract.gharchive --hour 2024-06-15-10
python -m transform.events \
  --input  data/raw/2024-06-15-10.json.gz \
  --output data/out/2024-06-15-10.csv
head -3 data/out/2024-06-15-10.csv

✓ Self-check: the new CSV is non-empty and PushEvent is the top row.

Step 7 — Look at the file sizes

Compare the raw archive, the compressed input, and the CSV output:

du -h data/raw/*.json.gz data/out/*.csv

The CSV is far smaller than the .json.gz — that's expected. We threw away 99% of the data: only the type field survived into the CSV, and we aggregated from hundreds of thousands of events down to roughly 30 event-type rows. Lesson 02 will keep more columns and still produce a smaller file — that's the power of columnar compression.

Troubleshooting