Foundations, week by week.

Stand up the course's foundational Git repository the way a real data team would: a Python virtual environment, one installed dependency pinned in requirements.txt, a tiny runnable script, a .gitignore that keeps secrets and the venv out of history, and a clear README. You practice the everyday DE workflow — feature branch, at least three small atomic commits, and a pull request you open, review, and merge — so the mechanics are muscle memory before any data is involved.

1. Create the repo skeleton locally: make the folder tree shown in the scaffold (src/, data/raw/, data/processed/, notes/, README.md, .gitignore, requirements.txt), then run `git init` and confirm `git status` shows the untracked files.
2. Create and activate a virtual environment (`python -m venv .venv` then `source .venv/bin/activate`, or the Windows equivalent), install ONE package (e.g. `pip install requests`), and freeze it with `pip freeze > requirements.txt`. Verify `.venv/` is ignored by confirming it does NOT appear in `git status`.
3. Write the small script `src/check_env.py` from the starter scaffold: complete the TODOs so it prints the Python version and confirms the installed package imports — this is your proof the environment is reproducible from requirements.txt alone.
4. Make your FIRST commit on `main` containing only the baseline: .gitignore, README skeleton, and requirements.txt (do not commit .venv or data files). Confirm with `git log --oneline` that exactly one commit exists.
5. Create and switch to a feature branch (`git switch -c feature/env-setup`), then build up the work in at least THREE atomic commits — e.g. (a) add `src/check_env.py`, (b) fill in the README run instructions, (c) add a `notes/decisions.md` entry. Each commit message states WHAT changed and WHY in the imperative mood.
6. Push the branch to GitHub (`git push -u origin feature/env-setup`) and open a Pull Request whose description explains what the repo is for and lists the acceptance items as a checklist a reviewer can tick off.
7. Self-review the PR diff in the GitHub UI as if you were a teammate: confirm no secrets, no .venv, and no large data files leaked into the diff; leave at least one review comment on your own PR, then merge it (squash or merge-commit, chosen deliberately) and delete the branch.
8. Prove idempotency of your setup: on a clean checkout (or after `git switch main && git pull`), recreate the venv from scratch and run `pip install -r requirements.txt` followed by `python src/check_env.py` — it must succeed with no manual fixes. Record this in `notes/decisions.md`.
9. Write a 4-6 line `notes/decisions.md` reflection: which files you deliberately kept OUT of Git and why, why you chose squash vs merge-commit, and what a teammate would need to reproduce your environment.

• Pro Git book (esp. ch. 2 Git Basics & ch. 3 Branching)
• GitHub Docs: About pull requests
• Python docs: venv — Creation of virtual environments
• Python Packaging: Installing packages with pip & requirements files

"""check_env.py — proves this repo's environment is reproducible.

Run after: python -m venv .venv && source .venv/bin/activate
           pip install -r requirements.txt
Usage:     python src/check_env.py

This is a STARTER SCAFFOLD. Fill in the TODOs — do not just run it as-is.
"""
import sys


def report_python_version() -> None:
    # Print the running interpreter version so a teammate can confirm
    # they're on a compatible Python (this course assumes 3.10+).
    print(f"Python: {sys.version.split()[0]}")
    # TODO: exit non-zero (sys.exit(1)) if the major.minor is below 3.10,
    #       so a broken env fails loudly instead of silently.


def check_dependency() -> None:
    # TODO: import the ONE package you installed and pinned in
    #       requirements.txt (e.g. `requests`). Wrap it in try/except
    #       ImportError and print a clear message telling the user to run
    #       `pip install -r requirements.txt` if the import fails.
    # TODO: print the package's __version__ so the output is proof the
    #       pinned dependency is actually present.
    pass


def main() -> None:
    report_python_version()
    check_dependency()
    # TODO: print a final one-line 'environment OK' banner ONLY if both
    #       checks passed — this line is what your README tells a reviewer
    #       to look for.


if __name__ == "__main__":
    main()

• If I Started Over As A Data Engineer
• The 2026 Data Engineer Roadmap
• Free DE Checklist (template)
• DE Honeycomb roadmap (free)

• Introduction to Git DataCamp · Course
• Introduction to Shell DataCamp · Course
• The Git & GitHub Bootcamp (Colt Steele) Udemy · Course

Pull a live sample of NYC 311 Service Requests from the Socrata API, inspect the real mess (drifting complaint types, null locations, late-closed tickets), then map the record's end-to-end lifecycle as a Mermaid diagram and write a one-page data contract that declares the grain, the required fields, the known failure modes, and the downstream consumers who break when the data drifts. Deliverable is documentation, not a pipeline — but every claim in it must be backed by something you actually observed in the sample.

1. Land the data: run the provided curl scaffold to pull ~50 live rows from https://data.cityofnewyork.us/resource/erm2-nwe9.json into sample_311.json, then open it and skim the raw JSON so you see the actual field names (e.g. unique_key, complaint_type, descriptor, borough, latitude, incident_zip, created_date, closed_date, status).
2. Profile the mess concretely: for your sample, count how many rows have a null borough, latitude, or incident_zip; list the distinct complaint_type values; and find at least one record that is 'closed' but has no closed_date (or a closed_date before created_date). Record the exact counts and one example unique_key for each problem — these become your evidence.
3. Identify the grain in one sentence: state what exactly one row represents (one service request identified by unique_key) and name the column that makes it unique. Verify it by checking that unique_key has no duplicates in your sample; note what you'd do if it did.
4. Map the lifecycle: starting from the Mermaid skeleton, draw the record's journey from citizen intake (call/app/web) through the 311 system, the Socrata/Open Data publication layer, ingestion, storage, and the two or three dashboards/consumers at the end. On the edges where the data can silently break, annotate the failure (e.g. 'complaint_type renamed upstream', 'null borough -> dropped by borough filter', 'late closed_date -> SLA metric wrong').
5. Write the one-page data contract (Markdown) with these sections: Grain; Source & refresh cadence; Required (non-null) fields and their types; Known failure modes with the evidence counts from step 2; Idempotency/rerun policy (state how a re-pull must not create duplicate rows — merge on unique_key); and Consumers (who reads it and what breaks for them when a field drifts or goes null).
6. Declare consumers explicitly: name at least two downstream consumers (e.g. 'crew-routing dashboard filters on borough + complaint_type', 'SLA report aggregates closed_date - created_date') and, for each, write the one sentence describing the concrete harm when the contract is violated.
7. Write a short 'why' note (3-5 sentences): defend one real decision you made — for example, why you chose unique_key as the merge key for idempotent reloads, or why you treat null borough as a quarantine-worthy reject rather than a silent drop. This note is your interview answer.
8. Assemble README.md containing the business framing, the embedded Mermaid diagram, the data contract, the consumer list, and the why-note; commit sample_311.json alongside it as the evidence the numbers came from.

• Socrata SODA API — Getting Started & query basics ($limit, $where, $select)
• Socrata SODA API — Filtering & querying data with SoQL
• Mermaid — Flowchart syntax (nodes, edges, edge labels)
• NYC 311 Service Requests — dataset page & column dictionary

#!/usr/bin/env bash
# Week 2 starter scaffold — LANDS the raw data only. You build the analysis + contract.
set -euo pipefail

# --- 1. Pull ~50 live rows from the NYC 311 Socrata endpoint ------------------
# $limit caps the rows. No app token is needed for small pulls (you may get
# throttled/429 if you hammer it — that throttling IS one of the messes you
# should note in your contract).
API="https://data.cityofnewyork.us/resource/erm2-nwe9.json"
curl -s "${API}?\$limit=50" -o sample_311.json
echo "Wrote sample_311.json ($(wc -c < sample_311.json) bytes)"

# TODO(you): profile the mess. Suggested one-liners (needs jq):
#   jq '[.[] | select(.borough==null)] | length' sample_311.json      # null boroughs
#   jq -r '[.[].complaint_type] | unique[]' sample_311.json           # distinct categories
#   jq '[.[] | select(.status=="Closed" and (.closed_date==null))]|length' sample_311.json
# Record the COUNTS + one example unique_key each — that is your evidence.

# TODO(you): confirm the grain — prove unique_key is unique in the sample:
#   jq -r '.[].unique_key' sample_311.json | sort | uniq -d   # should print nothing

# --- 2. Mermaid lifecycle SKELETON — copy into your README and finish it ------
# Replace the ??? edge labels with the SPECIFIC failure you observed, and add
# the real downstream consumers at the end.
cat <<'MERMAID'
```mermaid
flowchart LR
  A[Citizen: call / app / web] --> B[NYC 311 intake system]
  B --> C[Socrata Open Data publish]
  C -->|??? category drift?| D[Ingest / pull]
  D -->|??? null borough?| E[(Storage table)]
  E --> F[Crew-routing dashboard]
  E --> G[SLA / closure-time report]
  %% TODO(you): add a quarantine path for rejected rows (reason + timestamp)
  %% TODO(you): label every edge where data can SILENTLY break
```
MERMAID

• The Last Day of the Month Is Why You Have This Job!

• Understanding Data Engineering DataCamp · Course
• Fundamentals of Data Engineering O'Reilly · Book
• Introduction to Data Engineering (DeepLearning.AI) Coursera · Course

Build quakestats.py, a command-line tool that reads the live USGS all_day earthquake CSV, filters events above a magnitude threshold passed on the command line, and prints a summary — while surviving blank/missing magnitudes and malformed place strings instead of crashing. You will structure the logic into parse, classify, and summarize functions and prove robustness by running it twice with different thresholds.

1. Fetch a snapshot: download the all_day CSV once to a local file (e.g. `curl -sL https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/all_day.csv -o all_day.csv`) so you develop against a stable copy, then open it and note the columns (time, latitude, longitude, depth, mag, place, ...) and confirm some `mag` fields are blank.
2. Parse argv: read the magnitude threshold from sys.argv[1], convert it to float, and exit with a clear usage message and non-zero status if it is missing or non-numeric (fail loud on operator error, not silently).
3. Write parse_row(row): given a csv.DictReader row, return a clean dict with a float magnitude and a safe place string — but when `mag` is blank or non-numeric, do NOT default it to 0.0; instead flag the row as unparseable and return it (or None) so the caller can quarantine it. Handle malformed/missing place with a fallback like 'unknown location'.
4. Quarantine bad records: collect every row that fails magnitude parsing into a rejected list with the reason (e.g. 'blank mag', 'non-numeric mag') and the event id, and write them to rejected.csv with a reason column — never let a bad row silently vanish or crash the loop.
5. Write classify(mag): using if/elif/else control flow, bucket a magnitude into a severity label (e.g. minor < 3.0, light 3.0–4.9, moderate 5.0–5.9, strong 6.0+); write summarize(events) to count how many valid events cleared the threshold, group counts by severity, and track the single strongest quake and its place.
6. Loop and filter: iterate the CSV with csv.DictReader, call parse_row on each, skip+quarantine failures, keep only events with mag >= threshold, and feed survivors to classify/summarize. Print a readable report: total rows read, count above threshold, counts per severity, the strongest event, and how many rows were rejected and why.
7. Run it twice against the same snapshot with different thresholds (e.g. `python quakestats.py 2.0` then `python quakestats.py 4.5`) and confirm the above-threshold count drops while the rejected count stays identical — the dirty-data handling is threshold-independent.
8. Write a 4–6 sentence NOTES.md: state exactly what you did with blank magnitudes and why (quarantine vs. drop vs. default-to-zero — and why defaulting to 0.0 would silently hide real events), and note that re-running overwrites rejected.csv rather than appending, so reruns stay idempotent.

• USGS Real-time GeoJSON/CSV Feeds — documentation & feed formats
• Python docs — csv module (DictReader, DictWriter)
• Python docs — sys.argv (command-line arguments)
• USGS — magnitude / event terminology reference

#!/usr/bin/env python3
"""quakestats.py — threshold alerting over the USGS all_day earthquake feed.

Usage:  python quakestats.py <magnitude_threshold>
Example: python quakestats.py 4.5
"""
import csv
import sys

FEED_URL = "https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/all_day.csv"
LOCAL_CSV = "all_day.csv"      # download once with curl, develop against this snapshot
REJECTED_CSV = "rejected.csv"  # bad rows land here with a reason


def parse_args(argv):
    """Return the float threshold from argv, or exit with a usage message."""
    # TODO: require exactly one argument; convert to float.
    # TODO: on missing/non-numeric input, print usage to stderr and sys.exit(1).
    ...


def read_rows(path):
    """Yield dict rows from the local CSV snapshot."""
    with open(path, newline="", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        for row in reader:
            yield row
    # NOTE: the live feed changes every minute — download once, then read the file.


def parse_row(row):
    """Return (event, reason). event is a clean dict, or None if unparseable.

    reason is None on success, else a short string like 'blank mag'.
    """
    # TODO: pull row['mag']; if blank or non-numeric, return (None, '<reason>').
    #       Do NOT default a missing magnitude to 0.0 — that hides real gaps.
    # TODO: build a clean dict: {'id':..., 'mag': float, 'place': ...}
    #       with a fallback place string when row['place'] is missing/malformed.
    ...


def classify(mag):
    """Return a severity label for a magnitude using if/elif/else."""
    # TODO: minor / light / moderate / strong buckets.
    ...


def summarize(events):
    """Return a summary of events that cleared the threshold."""
    # TODO: total count, counts per severity label, and the strongest event.
    ...


def main():
    threshold = parse_args(sys.argv)
    kept, rejected = [], []

    for row in read_rows(LOCAL_CSV):
        event, reason = parse_row(row)
        if event is None:
            # TODO: append (id, reason) to rejected — quarantine, don't crash.
            continue
        # TODO: keep only events with event['mag'] >= threshold.

    # TODO: write rejected rows to REJECTED_CSV (overwrite, not append -> idempotent).
    # TODO: call summarize(kept) and print a readable report:
    #       rows read, count above threshold, per-severity counts,
    #       strongest event + place, and rejected count with reasons.


if __name__ == "__main__":
    main()

• The Most Expensive Python Mistakes You're Making
• Better-Python template (free)

• Introduction to Python DataCamp · Course
• The Complete Python Bootcamp From Zero to Hero (Jose Portilla) Udemy · Course
• Python for Everybody Specialization (U. Michigan) Coursera · Specialization

Ingest the live USGS all_month earthquake GeoJSON, iterate its nested features, and for each one validate/cast the fields, dedupe by event id with a set, and aggregate event counts and max magnitude per region into a dict. Every record that fails a check is written to a quarantine file with a specific reason and a timestamp rather than crashing the run — then you deliberately inject malformed records and prove the run finishes clean.

1. Download the feed: fetch https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/all_month.geojson and save it locally as all_month.geojson so your run is reproducible and you are not re-hitting the API on every rerun; open it and note the top-level 'features' array and the nested 'properties'/'geometry' shape.
2. Load the file with the json module inside a try/except and iterate the 'features' list; for each feature pull id, properties.mag, properties.place, and geometry.coordinates, wrapping the access so a truncated/missing-key object raises a caught exception instead of stopping the loop.
3. Write validation rules that decide good vs. bad: id must be present and non-empty; mag must be castable to float and non-null; place must be a non-empty string; coordinates must be a 3-element list. Each failing rule produces a specific reason string (e.g. 'null_mag', 'missing_id', 'bad_coordinates').
4. Dedupe by id using a set of seen ids: the first time you see an id you process it, and any later feature with the same id is skipped (or sent to quarantine with reason 'duplicate_id') so overlapping pulls never double-count.
5. Derive a region key from place (e.g. the text after the last comma, or 'unknown' when place is null) and aggregate into a dict per region: increment an event count and track the running max magnitude.
6. For every rejected record, call your quarantine writer to append one JSON line to rejects.jsonl containing the raw record (or its id), the reason, and a UTC timestamp — so a reviewer can see what was dropped and why.
7. Inject failure on purpose: append 3–4 hand-crafted malformed features to a copy of the input (a null mag, a missing id, a truncated object cut mid-field, a duplicate id) and rerun; confirm the program finishes without crashing and each bad record lands in rejects.jsonl with the right reason.
8. Prove idempotency: run the whole script twice against the same input and show the region aggregates are identical and rejects.jsonl is not double-appended — pick and document your strategy (truncate/overwrite the rejects file at start of run, and dedupe by id so counts don't grow).
9. Write a short DECISIONS.md: state your validation rules, why bad records are quarantined rather than dropped, your dedupe key, and your rerun strategy — these are your interview answers.

• GeoJSON format specification (geojson.org)
• Python docs — json module
• Python docs — Errors and Exceptions (try/except)
• USGS Earthquake GeoJSON feed — field/format documentation

"""Week 4 lab scaffold: ingest USGS all_month GeoJSON, quarantine bad records.
Stdlib only. Fill in every TODO. This is a starting structure, NOT a solution.
"""
import json
import datetime
import urllib.request

FEED_URL = "https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/all_month.geojson"
INPUT_FILE = "all_month.geojson"
REJECTS_FILE = "rejects.jsonl"


def download_feed(url=FEED_URL, dest=INPUT_FILE):
    """Land the raw feed locally so reruns don't re-hit the API."""
    urllib.request.urlretrieve(url, dest)
    print(f"saved feed -> {dest}")


def quarantine(record, reason, rejects_path=REJECTS_FILE):
    """Append one bad record to the rejects file with a reason + UTC timestamp.
    STUB: keep this shape. A reviewer must be able to answer 'what was dropped and why?'
    """
    row = {
        "ts": datetime.datetime.now(datetime.timezone.utc).isoformat(),
        "reason": reason,
        "record": record,  # TODO: consider storing just the id if records are large
    }
    with open(rejects_path, "a", encoding="utf-8") as f:
        f.write(json.dumps(row) + "\n")


def main():
    # download_feed()  # uncomment once, then work off the local file

    # TODO: decide your idempotent-rerun strategy. e.g. truncate REJECTS_FILE here
    #       so a second run doesn't append duplicate reject rows.

    with open(INPUT_FILE, encoding="utf-8") as f:
        data = json.load(f)

    seen_ids = set()          # TODO: use this to dedupe by event id
    regions = {}              # TODO: {region_key: {"count": int, "max_mag": float}}

    for feature in data.get("features", []):
        try:
            # TODO: safely pull id, properties.mag, properties.place,
            #       geometry.coordinates. A truncated object should raise here
            #       and be caught below -> quarantine, NOT crash.

            # TODO: validation rules. On each failure call:
            #       quarantine(feature, "<specific_reason>"); then skip this feature.
            #       reasons to cover: missing_id, null_mag, bad_mag_cast,
            #       empty_place, bad_coordinates, duplicate_id

            # TODO: dedupe with seen_ids

            # TODO: derive region key from place, then aggregate count + max_mag
            pass
        except Exception as e:
            quarantine(feature, f"parse_error:{type(e).__name__}")
            continue

    # TODO: print the per-region summary (count + max magnitude) sorted by count


if __name__ == "__main__":
    main()

• The Most Expensive Python Mistakes You're Making
• Better-Python template (free)

• Intermediate Python DataCamp · Course
• Python Crash Course, 3rd Ed. O'Reilly · Book
• The Complete Python Bootcamp From Zero to Hero (Jose Portilla) Udemy · Course

Refactor your Week 3–4 USGS earthquake scripts into a proper installable Python package, quakes/, with a QuakeClient class (client.py), pure parsing functions (parse.py), a report writer (report.py), and a CLI entry point (__main__.py). You'll create a fresh virtual environment, write a minimal pyproject.toml, install the package editable with pip install -e ., and prove it works end-to-end via both import quakes and python -m quakes.

1. Create the target package tree exactly as scaffolded: a project root holding pyproject.toml, README.md, and a quakes/ package directory containing __init__.py, client.py, parse.py, report.py, and __main__.py. Confirm quakes/ has an __init__.py so Python treats it as a real package, not a loose folder of scripts.
2. Create and activate a FRESH virtual environment at the project root (python -m venv .venv then source .venv/bin/activate on macOS/Linux or .venv\Scripts\activate on Windows). Confirm you are isolated by running which python (or where python) and pip list — the list should be nearly empty. Add .venv/ to .gitignore.
3. Fill in pyproject.toml from the scaffold: set name, version, requires-python, dependencies (requests), and a [project.scripts] entry so quakes maps to quakes.__main__:main. Read the Packaging User Guide section on pyproject.toml before choosing setuptools vs hatchling as your build backend, and write one sentence in your notes on why you picked it.
4. Refactor your Week 3–4 fetch logic into the QuakeClient class in client.py: the class holds the feed URL and a requests.Session, and exposes a fetch_all_day() method that returns the parsed GeoJSON dict. Move the fielding-out logic (pulling id, place, mag, time from each feature) into pure, side-effect-free functions in parse.py that take data in and return rows out — no network, no printing.
5. Wire report.py to take the parsed rows and write a deterministic output (CSV or a summary text file), and wire __main__.py to define main() that instantiates QuakeClient, calls fetch, passes results through parse, then report. Make the run idempotent: writing to a fixed output path must OVERWRITE, not append, so re-running python -m quakes twice produces one identical file, not duplicated rows.
6. Install the package in editable mode from the project root: pip install -e . . Confirm it landed by running pip list (you should see quakes) and pip show quakes.
7. Prove importability two ways: (a) open a Python REPL and run import quakes; from quakes.client import QuakeClient and confirm no ImportError; (b) run python -m quakes from a DIFFERENT working directory (e.g. your home folder) and confirm it still fetches and writes the report — this proves the package is installed, not just found because you happened to be in its folder.
8. Write a short DECISIONS.md (5–8 lines) answering: which build backend you chose and why, why the network code lives in a class but parsing lives in plain functions (testability), and how you made reruns idempotent. Commit the whole package to a GitHub repo with a README showing the install + run commands.

• Python Packaging User Guide — Packaging Python Projects tutorial
• Writing your pyproject.toml (setuptools & hatchling build backends)
• Python docs — Modules and Packages
• pip — Editable installs (pip install -e .)

# ============================================================
# STARTER SCAFFOLD ONLY — sets up the package skeleton.
# You write the real fetch / parse / report logic (see TODOs).
# ============================================================

# ---- Target package tree (create these files) ----
# quakes-project/
# |- pyproject.toml
# |- README.md
# |- .gitignore          # add: .venv/
# `- quakes/
#    |- __init__.py       # can be empty (marks it a package)
#    |- client.py         # QuakeClient class (network lives here)
#    |- parse.py          # PURE functions (no network, no print)
#    |- report.py         # write rows -> file (overwrite = idempotent)
#    `- __main__.py       # main(): wires client -> parse -> report


# ================= pyproject.toml =================
# [build-system]
# requires = ["setuptools>=68"]        # TODO: or swap to hatchling
# build-backend = "setuptools.build_meta"
#
# [project]
# name = "quakes"
# version = "0.1.0"
# requires-python = ">=3.9"
# dependencies = ["requests"]
#
# [project.scripts]
# quakes = "quakes.__main__:main"      # lets `quakes` run from shell too


# ================= quakes/client.py =================
import requests

ALL_DAY_URL = (
    "https://earthquake.usgs.gov/earthquakes/feed/"
    "v1.0/summary/all_day.geojson"
)

class QuakeClient:
    """Owns the feed URL + an HTTP session. Network belongs HERE only."""

    def __init__(self, url=ALL_DAY_URL, session=None):
        self.url = url
        self.session = session or requests.Session()

    def fetch_all_day(self):
        # TODO: GET self.url, raise_for_status(), return response.json()
        # TODO: (reuse your Week 3-4 error handling / timeout here)
        raise NotImplementedError


# ================= quakes/parse.py =================
def feature_to_row(feature):
    """PURE: one GeoJSON feature dict -> a flat dict. No I/O."""
    # TODO: pull id, place, mag, time from feature['properties'] / ['id']
    raise NotImplementedError

def rows_from_geojson(data):
    """PURE: full feed dict -> list of rows. No network, no printing."""
    # TODO: iterate data['features'], call feature_to_row on each
    raise NotImplementedError


# ================= quakes/report.py =================
def write_report(rows, path="quakes_report.csv"):
    # TODO: OVERWRITE path (mode 'w') so a rerun does NOT duplicate rows
    # TODO: write a header + one line per row (csv module is fine)
    raise NotImplementedError


# ================= quakes/__main__.py =================
from quakes.client import QuakeClient
from quakes.parse import rows_from_geojson
from quakes.report import write_report

def main():
    # TODO: client = QuakeClient(); data = client.fetch_all_day()
    # TODO: rows = rows_from_geojson(data); write_report(rows)
    # TODO: print a one-line summary (count of quakes written)
    raise NotImplementedError

if __name__ == "__main__":
    main()

• The Most Expensive Python Mistakes You're Making
• Better-Python template (free)

• Developing Python Packages DataCamp · Course
• Introduction to Object-Oriented Programming in Python DataCamp · Course
• Fluent Python, 2nd Ed. O'Reilly · Book

Pull one real month of NYC 311 service requests via Socrata, profile its dtypes/nulls/duplicate keys, write and execute a short cleaning plan (normalize borough casing, parse created_date, quarantine rejects), aggregate to complaint_type x borough x month, then merge to a tiny borough-population reference to compute per-10k rates. Prove the merge did not inflate rows (output row count == grain cardinality) and export a validated parquet.

1. LAND the raw data: query the Socrata endpoint for exactly one month using a $where on created_date (e.g. created_date between '2023-06-01' and '2023-07-01') plus $limit=200000, and save the CSV locally so reruns don't re-hit the API.
2. PROFILE the raw frame: print shape, df.dtypes, per-column null counts (df.isna().sum()), unique_key duplicate count (df['unique_key'].duplicated().sum()), and value_counts() on the borough column so you SEE the mixed casing ('BROOKLYN', 'Brooklyn', 'Unspecified') and datetime stored as object.
3. WRITE a cleaning plan as a short markdown/comment block BEFORE coding it: which columns you keep, how you parse created_date (pd.to_datetime, errors='coerce'), how you normalize borough (strip + str.upper or title-case to one canonical form), your rule for rows failing the id/date null check, and your dedup rule on unique_key (keep first). Decisions become interview answers.
4. EXECUTE cleaning and QUARANTINE bad records: send rows that fail the checks (null unique_key, unparseable created_date, borough not in the 5 canonical boroughs / 'Unspecified') to a rejected DataFrame with a reject_reason column and a rejected_at timestamp, write it to rejected_311.parquet, and keep only clean rows in the main frame.
5. TRANSFORM + AGGREGATE: derive a month period column from created_date, then groupby(['complaint_type','borough','month']).size() (or .agg) to build the request_count fact at that exact grain; confirm the grain is unique with an assertion on the groupby keys.
6. PREP + INSPECT the population reference: load the borough population table, normalize its borough key with the SAME normalization function, and PROVE the join key is unique (assert pop['borough'].is_unique) — this is the guard against silent row multiplication.
7. MERGE and VALIDATE: left-merge the aggregate to population on the normalized borough key, then assert the merged row count == the pre-merge aggregate row count (no inflation) and assert there are no null population values (every borough matched). Compute requests_per_10k = request_count / population * 10000.
8. MAKE THE RERUN IDEMPOTENT and EXPORT: write the final table to parquet with overwrite semantics (or partition by month) so a second run reproduces the same file instead of appending duplicates, and export requests_per_10k_by_type_borough_month.parquet.
9. WRITE the why-note (5–8 sentences): state your normalization choice, your reject rule, and specifically how you detected/prevented merge inflation (the row-count-equality and unique-key assertions), plus what the rate would have looked like if the key had been non-unique.

• pandas User Guide — Merge, join, concatenate and compare
• pandas User Guide — Working with missing data
• Socrata SoQL — $where and $limit query parameters
• pandas.DataFrame.merge — validate and indicator parameters

import pandas as pd

# --- 1. LAND one month of raw 311 (scope with a SoQL $where so you don't pull everything) ---
START, END = "2023-06-01", "2023-07-01"  # TODO: pick your month
BASE = "https://data.cityofnewyork.us/resource/erm2-nwe9.csv"
WHERE = f"created_date >= '{START}' AND created_date < '{END}'"
URL = f"{BASE}?$limit=200000&$where={WHERE}"
# Tip: download once, then read the local file on reruns so you don't re-hit the API (429s).
raw = pd.read_csv(URL, low_memory=False)
# raw.to_csv('raw_311_month.csv', index=False)  # cache it


def profile(df):
    """Print the mess so your cleaning plan is evidence-based."""
    print("shape:", df.shape)
    print("\ndtypes:\n", df.dtypes)
    print("\nnull counts:\n", df.isna().sum().sort_values(ascending=False).head(15))
    print("\nduplicate unique_key rows:", df['unique_key'].duplicated().sum())
    print("\nborough value_counts (look at the CASING):\n", df['borough'].value_counts(dropna=False))
    # TODO: also inspect created_date dtype — is it object (string) or datetime?


profile(raw)

# --- 2. TODO: write your cleaning plan (comment block), then execute it ---
# TODO: parse created_date with pd.to_datetime(..., errors='coerce')
# TODO: normalize borough to ONE canonical form (define borough_norm() and reuse it on BOTH frames)
# TODO: quarantine rows failing null/date/borough checks into a rejected frame with
#       reject_reason + rejected_at columns, write rejected_311.parquet, keep clean rows only
# TODO: dedup on unique_key (decide keep='first' and say why in the note)
# TODO: derive month, then groupby(['complaint_type','borough','month']).size() -> agg fact


def assert_no_merge_inflation(left, merged, on):
    """The validation IS the deliverable. Call this right after your merge."""
    # A unique key on the RIGHT table is what prevents row multiplication.
    assert len(merged) == len(left), (
        f"MERGE INFLATED ROWS: left={len(left)} -> merged={len(merged)}. "
        f"Your right-side '{on}' key is probably non-unique."
    )
    # TODO: also assert every left row matched (no null population after merge)
    # TODO: also assert pop_ref[on].is_unique BEFORE you merge
    print("merge OK: no inflation, grain preserved")


# --- 3. TODO: load population ref, normalize its borough with the SAME borough_norm(),
#             assert it's unique, left-merge, call assert_no_merge_inflation(),
#             compute requests_per_10k, and export the parquet idempotently (overwrite). ---

• Big Data Kills Pandas Performance | Use This Instead
• PySpark vs Pandas - Which Gets You Hired?
• DataFrame Upgrade Roadmap (free)

• Data Manipulation with pandas DataCamp · Course
• Python for Data Analysis, 3rd Ed. O'Reilly · Book
• Data Analysis with Pandas and Python (Boris Paskhaver) Udemy · Course

Paginate the World Bank population API (SP.POP.TOTL) with exponential backoff on 429/timeouts, normalize its nested JSON into rows keyed by ISO3+year, load the OWID population CSV, then outer-join the two and classify every key as matched / api-only / csv-only. Compare overlapping values against a tolerance to flag agreements vs. real discrepancies, and emit one reconciled table where every row carries source and as-of provenance.

1. Set up: create a fresh virtualenv, `pip install requests pandas`, and read the three reference docs (World Bank Developer Information, requests timeouts/raise_for_status, MDN 429/Retry-After) so you know the pagination contract and the retry rules before writing code.
2. Build `paginate()`: call the World Bank endpoint, read `page`, `pages`, and `total` from the metadata object at `response[0]` (element 0 is the header, element 1 is the data array), and loop `page=1..pages`. Wrap each request with a connect+read `timeout` and `raise_for_status()`; on a 429 or timeout, back off exponentially (e.g. 1s, 2s, 4s, capped) and honor the `Retry-After` header when present, giving up after N tries.
3. Normalize the API payload into flat records keyed by ISO3 + year: pull `countryiso3code`, `date` (year), and `value`. Drop rows where `countryiso3code` is empty (World Bank returns aggregates like regions with blank/2-letter codes) and quarantine rows where `value` is null into a `rejected` list with a reason ('null_value') and a UTC `rejected_at` timestamp rather than silently dropping them.
4. Load the OWID population CSV with pandas. Reconcile its `Entity` names / `Code` column to ISO3: keep rows that have a real 3-letter ISO3 `Code`, and quarantine the aggregate 'regions' (e.g. 'World', 'Africa', income groups whose Code is blank or non-ISO3) to the same rejected table with reason 'not_a_country' — do NOT let them collide with country keys.
5. Implement `reconcile(api_df, csv_df)`: outer-join on (iso3, year) and label each key `matched` (in both), `api_only` (World Bank has it, OWID doesn't), or `csv_only` (OWID has it, World Bank doesn't). Return counts per class so you can prove the inner join would have hidden the api_only + csv_only rows.
6. For `matched` keys, compare the two population values with a relative tolerance (e.g. abs(a-b)/max(a,b) <= 0.005 = 0.5%). Add a `discrepancy` flag and a `pct_diff` column so within-tolerance agreements are distinguished from genuine disagreements you must explain.
7. Write ONE reconciled table (Parquet or CSV) with columns: iso3, year, wb_value, owid_value, match_class, pct_diff, discrepancy, plus provenance columns `source_wb`, `source_owid` (the URLs), and `as_of` (UTC run timestamp). Make the write idempotent — overwrite the output file (or replace by run key) so a rerun does not append duplicates.
8. Write a short DECISIONS.md: your tolerance and why, how you keyed and why blank ISO3 rows are rejected not dropped, your backoff/Retry-After policy, and 2–3 concrete discrepancies you found (country+year+pct_diff) with a hypothesis for why the sources disagree.

• World Bank Indicators API — Developer Information (Basic Call Structures, pagination, format=json)
• requests — timeouts and Response.raise_for_status()
• MDN — HTTP 429 Too Many Requests & Retry-After header
• pandas — merge/join (how='outer', indicator=True)

import time
from datetime import datetime, timezone
import requests
import pandas as pd

WB_URL = "https://api.worldbank.org/v2/country/all/indicator/SP.POP.TOTL"
AS_OF = datetime.now(timezone.utc).isoformat()


def paginate(per_page=1000, max_retries=5, timeout=(5, 30)):
    """Yield the data array from every page, backing off on 429/timeouts."""
    page = 1
    while True:
        params = {"format": "json", "per_page": per_page, "page": page}
        delay = 1.0
        for attempt in range(max_retries):
            try:
                r = requests.get(WB_URL, params=params, timeout=timeout)
                if r.status_code == 429:
                    # TODO: honor Retry-After header if present, else use `delay`
                    time.sleep(delay); delay *= 2; continue
                r.raise_for_status()
                break
            except (requests.Timeout, requests.ConnectionError):
                if attempt == max_retries - 1:
                    raise
                time.sleep(delay); delay *= 2  # exponential backoff
        meta, data = r.json()  # element[0]=metadata, element[1]=rows
        yield data
        if page >= meta["pages"]:  # meta also has total/per_page
            break
        page += 1


def normalize_wb(pages):
    """TODO: flatten rows to dicts keyed by (iso3, year) with value.
    - use countryiso3code + date + value
    - DROP blank/non-ISO3 codes (aggregates)
    - QUARANTINE null value -> rejected list w/ reason + rejected_at (UTC)
    Return (clean_df, rejected_df)."""
    raise NotImplementedError


def load_owid(csv_path):
    """TODO: read population.csv; keep real ISO3 `Code` rows;
    quarantine aggregates (blank/non-ISO3 Code) with reason 'not_a_country'."""
    raise NotImplementedError


def reconcile(api_df, csv_df, tol=0.005):
    """Outer-join and classify each (iso3, year) key."""
    merged = api_df.merge(csv_df, on=["iso3", "year"], how="outer", indicator=True)
    merged["match_class"] = merged["_merge"].map({
        "both": "matched", "left_only": "api_only", "right_only": "csv_only"})
    # TODO: for matched rows compute pct_diff and set discrepancy = pct_diff > tol
    # TODO: attach provenance columns source_wb, source_owid, as_of = AS_OF
    return merged


if __name__ == "__main__":
    pages = list(paginate())
    # api_clean, api_rejected = normalize_wb(pages)
    # owid_clean, owid_rejected = load_owid("population.csv")
    # out = reconcile(api_clean, owid_clean)
    # TODO: idempotent write (overwrite) of reconciled + rejected tables
    print("pages fetched:", len(pages))

• Big Data Kills Pandas Performance | Use This Instead
• PySpark vs Pandas - Which Gets You Hired?
• DataFrame Upgrade Roadmap (free)

• Introduction to APIs in Python DataCamp · Course
• Using Python to Access Web Data (U. Michigan) Coursera · Course
• Joining Data with pandas DataCamp · Course

Harden your Week 7 USGS earthquake reconciliation into a proper src-layout Python package with clean io/reconcile/cli seams, full type hints that pass mypy with zero errors, and a pytest suite (parametrized over a committed GeoJSON fixture) covering the ugly cases — empty feed, duplicate ids across overlapping pulls, a missing field, and a metadata.count vs. actual-feature-count mismatch. Then build a wheel, install it into a clean venv, and run the whole thing through a console entry point exactly as a stranger cloning your repo would.

1. Capture the raw feed once and commit it as a frozen fixture: download https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/all_day.geojson to tests/fixtures/all_day.geojson, and make a second edited copy (tests/fixtures/all_day_overlap.geojson) that repeats 2-3 feature ids from the first pull so you have a real duplicate-id scenario committed to the repo (never fetch the network inside a test).
2. Scaffold the package with a src/ layout: create src/quake_recon/{__init__.py, io.py, reconcile.py, cli.py} and a pyproject.toml. Move your W7 logic into seams — io.py only loads/parses GeoJSON (and writes the rejected file), reconcile.py holds pure functions (no I/O), cli.py wires argparse to those functions and exposes a main().
3. Write the reconcile contract as typed pure functions: e.g. reconcile(features: list[Feature], reported_count: int) -> ReconcileResult that returns the actual feature count, the metadata.count delta, and a list of duplicate ids. Add type hints everywhere (use TypedDict or a dataclass for the GeoJSON feature shape) and run `mypy src/` until it reports zero errors.
4. Implement build-for-failure behavior: dedupe features on the feature id, and route any feature failing a null-id check to a rejected table/file (data/rejected/*.jsonl) with a `reason` column and an ISO `rejected_at` timestamp. Make the run idempotent — a rerun over the same pulls must merge/dedupe on id, not append duplicates.
5. Write a parametrized pytest suite in tests/test_reconcile.py that loads the committed fixtures and covers, as separate cases: empty feed (zero features), duplicate ids across the two overlapping pulls, a feature with a missing/null id field, and a metadata.count vs. actual-count mismatch — asserting the reconcile result and that bad records land in the rejected file with a reason.
6. Add a console_scripts entry point in pyproject.toml (e.g. `quake-recon = quake_recon.cli:main`) so the tool runs as a command, not `python somefile.py`. Run `pytest` and `mypy` together and get both green.
7. Build the distributable: `python -m build` to produce a wheel in dist/. Confirm the wheel filename and that it includes your package (not the tests).
8. Create a clean throwaway venv, `pip install dist/quake_recon-*.whl` into it (nothing else from your source tree on the path), and run `quake-recon --input tests/fixtures/all_day.geojson` via the entry point to prove it works as an installed component.
9. Write a short README contract (inputs, outputs, exit codes, where rejected records go) plus a WHY.md decisions note answering: why dedupe on id vs. drop, why null-id rows are quarantined instead of failing the whole run, and why you chose overwrite/merge for idempotency — these are your interview answers.

• pytest — fixtures and parametrize (how to use)
• pytest — parametrizing tests
• mypy — getting started with static typing
• Python Packaging — entry points / console_scripts

# ============================================================
# FILE: pyproject.toml
# ============================================================
# [build-system]
# requires = ["setuptools>=68"]
# build-backend = "setuptools.build_meta"
#
# [project]
# name = "quake_recon"
# version = "0.1.0"
# requires-python = ">=3.10"
#
# [project.scripts]
# quake-recon = "quake_recon.cli:main"   # <- console entry point
#
# [tool.setuptools.packages.find]
# where = ["src"]


# ============================================================
# FILE: src/quake_recon/io.py   (I/O seam — only place that touches disk)
# ============================================================
from __future__ import annotations
import json
from pathlib import Path
from typing import TypedDict


class Feature(TypedDict):
    id: str | None          # USGS feature id; can be missing/null in the wild
    properties: dict
    geometry: dict


def load_feed(path: Path) -> tuple[list[Feature], int]:
    """Return (features, reported_count) from a committed GeoJSON file.
    NOTE: read from disk only. Tests must never hit the network.
    """
    data = json.loads(path.read_text())
    reported_count = data["metadata"]["count"]  # may disagree with len(features)!
    features: list[Feature] = data["features"]
    return features, reported_count


def write_rejected(rows: list[dict], out: Path) -> None:
    # TODO: append each rejected row as JSONL with a `reason` and ISO `rejected_at`.
    # TODO: make this idempotent-safe (dedupe on id) so reruns don't pile up dups.
    raise NotImplementedError


# ============================================================
# FILE: src/quake_recon/reconcile.py   (pure logic — NO I/O here)
# ============================================================
from __future__ import annotations
from dataclasses import dataclass, field
from .io import Feature


@dataclass
class ReconcileResult:
    actual_count: int
    reported_count: int
    count_delta: int                       # reported - actual
    duplicate_ids: list[str] = field(default_factory=list)
    rejected: list[dict] = field(default_factory=list)  # null/missing-id rows + reason


def reconcile(features: list[Feature], reported_count: int) -> ReconcileResult:
    """THE CONTRACT. Given features + the header count, return a ReconcileResult.
    TODO: count actual features.
    TODO: find ids that appear more than once (the overlapping-pull dup case).
    TODO: quarantine features whose id is None/missing into `rejected` with a reason.
    TODO: compute count_delta = reported_count - actual_count.
    Keep this function pure: no printing, no file writes.
    """
    raise NotImplementedError


# ============================================================
# FILE: src/quake_recon/cli.py   (wiring seam — argparse -> functions)
# ============================================================
from __future__ import annotations
import argparse
from pathlib import Path
from .io import load_feed
from .reconcile import reconcile


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(prog="quake-recon")
    parser.add_argument("--input", required=True, type=Path)
    args = parser.parse_args(argv)

    features, reported = load_feed(args.input)
    result = reconcile(features, reported)
    # TODO: print a human-readable summary (actual vs reported, dup count).
    # TODO: write rejected rows via io.write_rejected(...).
    # TODO: return a non-zero exit code if count_delta != 0 (documented contract).
    return 0


if __name__ == "__main__":
    raise SystemExit(main())


# ============================================================
# FILE: tests/test_reconcile.py   (ONE parametrized example — add the rest)
# ============================================================
import json
from pathlib import Path
import pytest
from quake_recon.reconcile import reconcile
from quake_recon.io import load_feed

FIXTURES = Path(__file__).parent / "fixtures"


@pytest.fixture
def all_day() -> tuple[list, int]:
    return load_feed(FIXTURES / "all_day.geojson")


# Starter parametrization skeleton. Fill in expected values for each edge case,
# and add cases for: empty feed, missing/null id, count mismatch.
@pytest.mark.parametrize(
    "fixture_name, expect_duplicate_ids",
    [
        ("all_day.geojson", False),          # baseline single pull
        ("all_day_overlap.geojson", True),   # TODO: overlapping pull w/ repeated ids
    ],
)
def test_duplicate_detection(fixture_name: str, expect_duplicate_ids: bool) -> None:
    features, reported = load_feed(FIXTURES / fixture_name)
    result = reconcile(features, reported)
    # TODO: assert bool(result.duplicate_ids) == expect_duplicate_ids
    # TODO: assert result.actual_count == <the number you counted by hand>
    raise NotImplementedError

• Your Code Works Perfectly? That's A Problem
• $180k Data Engineers Don't Make These Python Mistakes

• Introduction to Testing in Python DataCamp · Course
• Robust Python O'Reilly · Book
• Python Testing with pytest, 2nd Ed. O'Reilly · Book

Load the Chinook sample database into DuckDB and write 8–10 documented SQL queries that exercise filtering, multi-hop INNER JOINs, a LEFT JOIN that surfaces parentless rows, an employee→manager self-join, and a fan-out that inflates a total. Chinook is intentionally clean so you can isolate join and grain mechanics; the point is to predict row counts before you run and explain every delta.

1. Install DuckDB (`pip install duckdb`), download `Chinook_Sqlite.sqlite` from the Chinook releases page, and run the starter scaffold to ATTACH the SQLite file and confirm the tables (Artist, Album, Track, Invoice, InvoiceLine, Customer, Employee) are visible.
2. Before writing joins, print the schema of the core tables and sketch the ERD on paper: mark each primary key and each foreign key (Album.ArtistId, Track.AlbumId, InvoiceLine.InvoiceId, Invoice.CustomerId, Customer.SupportRepId, Employee.ReportsTo). You will predict row counts from this sketch.
3. Write 2 FILTER queries: customers in one Country sorted by name, and tracks whose Name matches a LIKE pattern with a price threshold. Add a one-line comment stating the grain (one row per ___).
4. Write the multi-hop INNER JOIN: Artist → Album → Track → InvoiceLine → Invoice → Customer to answer 'which customers bought music by a given artist?'. Comment which rows this silently drops (artists never purchased, customers with no invoices).
5. Write the LEFT JOIN Artist → Album filtered to `WHERE Album.AlbumId IS NULL` to surface parentless rows — artists with zero albums — the exact rows the INNER version hid.
6. Write the self-join Employee e → Employee m ON `e.ReportsTo = m.EmployeeId`; use a LEFT join so the top manager (ReportsTo IS NULL) is not dropped, and output each employee next to their manager's name.
7. Demonstrate fan-out: join Invoice to InvoiceLine, show that one invoice becomes many rows, and prove that naively SUM-ing Invoice.Total after the join inflates revenue; then write the correct grain-aware aggregation.
8. Pick one pair of tables and write the SAME query twice — once INNER, once LEFT — capture both row counts, and write a 3–5 sentence note explaining the exact delta and which real report bug each version would cause.
9. Save each query as a numbered .sql file with a header comment stating its result grain and expected vs actual row count; commit the folder plus the WHY note as your deliverable.

• DuckDB — SQL Introduction
• DuckDB — SQLite extension (ATTACH / read .sqlite)
• DuckDB — Reading and writing Parquet files
• Chinook database schema & ERD (lerocha/chinook-database)

"""
Week 9 starter scaffold — LAND the Chinook data in DuckDB, then STOP.
This file only opens a connection and exposes the tables. The queries are YOUR job.

Setup (once):
    pip install duckdb
    # Download Chinook_Sqlite.sqlite from:
    #   https://github.com/lerocha/chinook-database/releases
    # and put it next to this file.
"""

import duckdb

DB_PATH = "chinook.duckdb"          # persistent DuckDB file (so your work survives reruns)
SQLITE_FILE = "Chinook_Sqlite.sqlite"

con = duckdb.connect(DB_PATH)

# The sqlite extension lets DuckDB read a .sqlite file directly.
con.execute("INSTALL sqlite; LOAD sqlite;")

# ATTACH mounts the SQLite DB read-only under the alias `chinook`.
# You can now query e.g.  chinook.Artist, chinook.Customer, chinook.Employee ...
con.execute(f"ATTACH '{SQLITE_FILE}' AS chinook (TYPE sqlite);")

# Sanity check — confirm tables are visible before you write real queries.
print(con.execute("SELECT name FROM (SHOW ALL TABLES) WHERE database = 'chinook'").fetchdf())

# ------------------------------------------------------------------
# TODO 1: Print the schema of Artist, Album, Track, InvoiceLine,
#         Invoice, Customer, Employee. Sketch the ERD before querying.
# TODO 2: FILTER — customers in a single Country, sorted; add a WHERE + LIKE.
# TODO 3: Multi-hop INNER JOIN artist -> album -> track -> invoiceline ->
#         invoice -> customer. Which customers bought a given artist?
# TODO 4: LEFT JOIN Artist -> Album to surface artists with NO albums
#         (the parentless rows: WHERE Album.AlbumId IS NULL).
# TODO 5: SELF-JOIN Employee e -> Employee m ON e.ReportsTo = m.EmployeeId.
#         Use LEFT so the top manager (ReportsTo IS NULL) is not dropped.
# TODO 6: FAN-OUT — join Invoice to InvoiceLine and show how one invoice
#         row becomes many; prove it inflates SUM(Total) if you double-count.
# TODO 7: Write ONE query twice (INNER vs LEFT on the same tables) and
#         record the row-count delta + WHY those rows differ.
# TODO 8: Save each query as a numbered .sql file with a comment header
#         stating the grain of the result (one row per ___).
# ------------------------------------------------------------------

con.close()

• Joining Data in SQL DataCamp · Course
• The Complete SQL Bootcamp (Jose Portilla) Udemy · Course
• Learning SQL, 3rd Ed. O'Reilly · Book

On the Chinook database in DuckDB, build a ~10-query analytical pack covering GROUP BY+HAVING, set operations (EXCEPT/INTERSECT), and scalar/EXISTS/derived-table subqueries. Every query is annotated with its exact result grain ("one row per country-genre pair") and a one-line correctness proof, and the pack includes a deliberate NOT IN + NULL trap that returns the wrong answer plus its NOT EXISTS fix.

1. Get the data: install the DuckDB CLI (or the Python `duckdb` package), clone https://github.com/lerocha/chinook-database, and load the Chinook schema into a local `chinook.duckdb` file — either by running the SQLite/SQL build script through DuckDB or by `INSTALL sqlite; LOAD sqlite; ATTACH 'Chinook_Sqlite.sqlite' AS chinook (TYPE sqlite);`. Run 2-3 `SELECT count(*)` checks against `invoices`, `invoice_items`/`InvoiceLine`, `tracks`, and `customers` and record the counts as your sanity baseline.
2. Write 3 GROUP BY + HAVING queries whose grain you can name in one sentence each: (a) total revenue per billing country, (b) revenue per genre, (c) customers per country having more than N invoices. For each, add a comment line stating the exact result grain — e.g. `-- grain: one row per BillingCountry; measure = SUM(Total)` — and confirm HAVING filters groups, not rows.
3. Prove your COUNT isn't lying about NULLs: pick a nullable column (e.g. `customers.Company` or `State`) and run `COUNT(*)` vs `COUNT(<column>)` side by side in one query. Comment the delta — `-- COUNT(*)=59, COUNT(State)=30 -> 29 customers have NULL State and are excluded by COUNT(col)` — so you can defend exactly what each number represents.
4. Write 2 set-operation queries against the same-shaped column lists: use EXCEPT to find customers who bought from genre A but never genre B, and INTERSECT to find tracks appearing in two different playlists (or customers active in two years). Note in a comment that set ops dedupe by default (vs UNION ALL) and why that matches the question's grain.
5. Write 3 subquery queries — one of each kind: a scalar subquery (e.g. each customer's spend vs the overall average in a `WHERE` or `SELECT`), an EXISTS/correlated subquery (customers who have at least one invoice over $X), and a derived-table subquery (aggregate in a subquery, then filter/join the result — e.g. per-customer totals wrapped as a derived table). Annotate the grain each one returns.
6. Build the NOT IN + NULL trap deliberately: write a `WHERE col NOT IN (SELECT nullable_col FROM ...)` where the subquery contains at least one NULL, and show it returns zero/too-few rows (three-valued logic makes the whole predicate never TRUE). Then write the correct version with `NOT EXISTS` (or add `WHERE nullable_col IS NOT NULL`) and show the row counts differ. Comment both with the row counts as evidence.
7. Assemble all queries into a single `analytics_pack.sql` file in run order, each block preceded by: a plain-English question, a `-- grain:` line, and a `-- why correct:` line. Ensure the file runs top-to-bottom idempotently (read-only SELECTs; if you materialize anything use `CREATE OR REPLACE VIEW`/`TABLE` so a rerun produces no duplicates and no errors).
8. Write a short `NOTES.md` (the interview-answer file): for the NOT IN trap, state the specific failure (NULL in the subquery -> NOT IN predicate evaluates to UNKNOWN -> row excluded) and your fix; and for one GROUP BY row, write the sentence you'd say to a stakeholder about exactly what that row counts and how NULLs were handled.

• DuckDB — Aggregate functions (COUNT, SUM, and NULL handling)
• DuckDB — GROUP BY and HAVING clauses
• DuckDB — Set operations (UNION / EXCEPT / INTERSECT)
• PostgreSQL — Subquery expressions (IN / NOT IN / EXISTS and the NULL pitfall)

• Intermediate SQL DataCamp · Course
• SQL for Data Science (UC Davis) Coursera · Course
• Learning SQL, 3rd Ed. O'Reilly · Book

Load a subset of NOAA GHCN-Daily observations for a handful of weather stations into DuckDB, quarantine bad readings (-9999 sentinels and failed QC flags) into a rejected table with reasons, then answer four business questions — top-N ranking per month, day-over-day change detection, a 7-day moving average, and a running cumulative total — each written as a single, named-CTE-structured query with no nested subquery pyramids.

1. Pick 4-6 stations and pull their data: either download a couple of per-station CSVs from the /access/ folder or grab a recent year file from /by_year/ (long format: ID, DATE, ELEMENT, VALUE, M-FLAG, Q-FLAG, S-FLAG, OBS-TIME). Note in your README exactly which stations/years you chose and why (e.g. mix of stations that start/stop reporting mid-period).
2. Start from the starter scaffold: use DuckDB read_csv to load the raw files into a `raw_obs` table, keeping only ELEMENT in ('TMAX','TMIN','PRCP') and casting DATE properly. Do NOT drop bad rows yet — you inspect them first.
3. Build a quarantine step: write rows to a `rejected_obs` table with a `reason` column and a `rejected_at` timestamp for (a) VALUE = -9999 sentinels and (b) any non-blank Q-FLAG (failed a QC check). Everything clean lands in a `clean_obs` table with VALUE scaled to real units (TMAX/TMIN are tenths of degrees C, PRCP is tenths of mm). Record the reject count per reason in your notes.
4. Q1 — Ranking: in ONE query using named CTEs, aggregate a monthly metric per station (e.g. total PRCP or avg TMAX per station-month) and rank top-N stations within each month. Choose RANK vs DENSE_RANK vs ROW_NUMBER deliberately and write one sentence defending your choice on ties (e.g. two stations with identical monthly rainfall).
5. Q2 — Period-over-period: in ONE CTE-structured query use LAG() over (PARTITION BY station, element ORDER BY date) to compute day-over-day change in TMAX and PRCP, then flag the single biggest swing per station. Handle the gap where a station skipped a day so the lag does not silently compare non-adjacent dates.
6. Q3 — Moving average: compute a 7-day moving average of TMAX per station using an explicit window frame `ROWS BETWEEN 6 PRECEDING AND CURRENT ROW`. In your notes, state what your frame does when fewer than 7 readings exist (leading edge / gaps) and why you chose ROWS over RANGE.
7. Q4 — Running total: compute a running cumulative PRCP total within each station partition (PARTITION BY station ORDER BY date, default frame) to model season-to-date accumulation. Confirm the total resets per station and never bleeds across partitions.
8. Make it idempotent: wrap the load in CREATE OR REPLACE TABLE (or DELETE-then-INSERT on the station+date+element key) so rerunning the whole script twice yields identical row counts — no duplicates. Verify with a COUNT(*) before and after a second run.
9. Write a short DECISIONS.md: which ranking function and why, how you defined an 'adjacent' day for LAG, your ROWS-frame edge behavior, your reject rules, and your idempotency strategy — these are your interview answers.

• GHCN-Daily documentation (readme, element codes, QC & source flags, units)
• DuckDB — Window Functions
• DuckDB — WITH clause (Common Table Expressions)
• DuckDB — read_csv / CSV import

-- Starter scaffold: land raw GHCN-Daily long-format data into DuckDB and
-- set up the normalize + quarantine split. This ONLY stages data.
-- You write the QC/quarantine logic and all four analytical queries.
--
-- Run with:  duckdb climate.duckdb < load_ghcn.sql
-- by_year files are long format with NO header, columns are:
--   ID, DATE(YYYYMMDD), ELEMENT, VALUE, M_FLAG, Q_FLAG, S_FLAG, OBS_TIME

-- 1) Land the raw rows. Point read_csv at your downloaded file(s).
--    (A glob like '2023.csv' or a couple of /access/ station files also works
--     but /access/ files are WIDE format -- adjust columns if you use those.)
CREATE OR REPLACE TABLE raw_obs AS
SELECT *
FROM read_csv(
    '2023.csv',                 -- TODO: your downloaded GHCN file(s) / glob
    header = false,
    columns = {
        'ID': 'VARCHAR',
        'OBS_DATE': 'VARCHAR',  -- keep as text for now, cast below
        'ELEMENT': 'VARCHAR',
        'VALUE': 'INTEGER',     -- tenths of units; -9999 = missing
        'M_FLAG': 'VARCHAR',
        'Q_FLAG': 'VARCHAR',    -- non-blank = failed a QC check
        'S_FLAG': 'VARCHAR',
        'OBS_TIME': 'VARCHAR'
    }
);

-- 2) Narrow to your stations + the 3 elements you care about.
--    TODO: replace with the station IDs you selected (see docs for ID format).
CREATE OR REPLACE VIEW obs_scoped AS
SELECT
    ID                                   AS station,
    strptime(OBS_DATE, '%Y%m%d')::DATE   AS obs_date,
    ELEMENT                              AS element,
    VALUE                               AS raw_value,
    Q_FLAG                               AS q_flag
FROM raw_obs
WHERE ELEMENT IN ('TMAX','TMIN','PRCP')
  AND ID IN ( 'USW00094728' /* TODO: add your 3-5 other station IDs */ );

-- 3) QUARANTINE: split bad rows out WITH A REASON + timestamp.
--    TODO: insert rows where raw_value = -9999 (reason 'missing_sentinel')
--          and rows where q_flag <> '' (reason 'failed_qc_' || q_flag)
--          into rejected_obs(station, obs_date, element, raw_value, reason, rejected_at).
CREATE OR REPLACE TABLE rejected_obs (
    station VARCHAR, obs_date DATE, element VARCHAR,
    raw_value INTEGER, reason VARCHAR, rejected_at TIMESTAMP
);
-- INSERT INTO rejected_obs SELECT ... ;   -- <-- your logic here

-- 4) CLEAN table: only rows that pass, with VALUE scaled to real units.
--    TMAX/TMIN are in tenths of degC; PRCP is in tenths of mm.
--    TODO: exclude anything you quarantined above, then divide by 10.0.
CREATE OR REPLACE TABLE clean_obs AS
SELECT
    station,
    obs_date,
    element,
    raw_value / 10.0 AS value      -- TODO: confirm scaling per element
FROM obs_scoped
WHERE 1 = 0;                       -- TODO: replace with your real WHERE (drop rejects)

-- 5) Sanity check before you build the 4 window queries.
SELECT 'clean' t, count(*) n FROM clean_obs
UNION ALL SELECT 'rejected', count(*) FROM rejected_obs;

-- ---- Your four named-CTE queries go in separate files (q1..q4.sql):
-- Q1 rank top-N stations per month  (defend RANK vs DENSE_RANK vs ROW_NUMBER)
-- Q2 LAG() day-over-day change + flag biggest swing (mind date gaps)
-- Q3 7-day moving avg with ROWS BETWEEN 6 PRECEDING AND CURRENT ROW
-- Q4 running cumulative PRCP within each station partition

• PostgreSQL Summary Stats and Window Functions DataCamp · Course
• SQL Cookbook, 2nd Ed. O'Reilly · Book

Load a multi-million-row slice of NOAA GHCN-Daily (one by_year file) into local PostgreSQL, then tune a provided slow date-range + station-filter aggregation: capture an EXPLAIN (ANALYZE, BUFFERS) baseline, pinpoint the bottleneck operator, rewrite the non-sargable date predicate, add one composed index, and prove each change against the plan — ending with a written decision note on the index you added and the one you deliberately did not.

1. 1. Download one by_year file from NOAA GHCN-Daily (e.g. 2020.csv.gz — tens of millions of global rows), gunzip it, and inspect the 8 comma-separated columns (ID, YYYYMMDD date, ELEMENT, VALUE, M/Q/S flags, OBS-TIME) — note there is no header row and the date arrives as an 8-digit string.
2. 2. Create the raw table and bulk-load with COPY exactly as in the scaffold, loading obs_date as TEXT on purpose (this is what forces the non-sargable predicate you will fix). Confirm the row count is in the millions with SELECT count(*).
3. 3. Run the provided intentionally-slow query and capture the baseline with EXPLAIN (ANALYZE, BUFFERS). Save the full plan text to a file (baseline_plan.txt) — record total time, the top-cost operator (expect a Parallel Seq Scan), rows removed by filter, and shared buffers hit/read.
4. 4. Diagnose why it is slow: identify that the Seq Scan reads every row and that to_date(obs_date,'YYYYMMDD') in the WHERE clause is non-sargable, so no index on the date could even be used. Write down the bottleneck operator and the reason in your decision note.
5. 5. Make the predicate sargable: rewrite the date filter to compare obs_date against literal text bounds ('20200601' AND '20200831') OR add a computed date column / expression index — choose one deliberately. Re-run EXPLAIN (ANALYZE, BUFFERS) and diff against baseline.
6. 6. Add ONE index that serves the filter + aggregation (compose station_id, element, and the sargable date column; consider column order and whether an index-only scan is achievable by INCLUDE-ing value). Run ANALYZE, re-EXPLAIN, and confirm the plan switched to an Index/Index-Only Scan with fewer buffers.
7. 7. Prove idempotence of your tuning artifacts: make the load rerunnable without duplicates (TRUNCATE-then-COPY, or COPY into a staging table then INSERT on a unique key), and guard index creation with CREATE INDEX IF NOT EXISTS so a rerun of your tuning script is safe.
8. 8. Write the decision note (tuning.md): the operator you fixed, the predicate rewrite, the index you added and its column order rationale, and — critically — one index you deliberately did NOT add (e.g. a single-column index on element with tiny cardinality, or a redundant index the planner would ignore) and why.

• PostgreSQL Documentation — Using EXPLAIN
• PostgreSQL Documentation — Indexes (B-tree, multicolumn, index-only scans)
• PostgreSQL Documentation — COPY (bulk load)
• PostgreSQL Documentation — Row Estimation & the Planner (why predicates must be sargable)

-- ============================================================
-- Week 12 STARTER SCAFFOLD — GHCN-Daily bulk load + slow query
-- Fills in the SETUP only. You build the tuning. Look for TODO.
-- ============================================================

-- 1) Raw table. NOTE: obs_date is TEXT on purpose (YYYYMMDD).
--    That choice is what makes the date filter non-sargable later.
CREATE TABLE IF NOT EXISTS ghcn_raw (
    station_id text,
    obs_date   text,      -- 8-digit YYYYMMDD string, e.g. '20200615'
    element    text,      -- 'TMAX','TMIN','PRCP',...
    value      integer,   -- tenths of a unit (deg C / mm)
    m_flag     text,
    q_flag     text,
    s_flag     text,
    obs_time   text
);

-- 2) Bulk load one by_year file with COPY (headerless CSV).
--    Download + gunzip first, e.g.:
--      curl -O https://www.ncei.noaa.gov/data/global-historical-climatology-network-daily/by_year/2020.csv.gz
--      gunzip 2020.csv.gz
--    IDEMPOTENCY: uncomment TRUNCATE so a rerun does NOT duplicate rows.
-- TRUNCATE ghcn_raw;
COPY ghcn_raw FROM '/absolute/path/to/2020.csv' WITH (FORMAT csv);

-- Sanity check: expect millions of rows.
-- SELECT count(*) FROM ghcn_raw;

ANALYZE ghcn_raw;

-- 3) THE INTENTIONALLY-SLOW QUERY. Do NOT edit it yet — baseline it first.
--    Summer daily-high averages for US stations.
SELECT station_id, element, avg(value) / 10.0 AS avg_deg_c
FROM   ghcn_raw
WHERE  to_date(obs_date, 'YYYYMMDD') BETWEEN DATE '2020-06-01' AND DATE '2020-08-31'  -- non-sargable!
  AND  station_id LIKE 'US%'
  AND  element = 'TMAX'
GROUP  BY station_id, element;

-- TODO(a): capture the baseline plan and save it:
--   EXPLAIN (ANALYZE, BUFFERS) <the query above>;
--   -> record total time, the top operator (Seq Scan?), rows removed by filter, buffers.

-- TODO(b): rewrite the date predicate to be SARGABLE.
--   Hint: obs_date is text in YYYYMMDD, which sorts lexically like dates.
--   Compare obs_date to text bounds instead of wrapping it in to_date(...).

-- TODO(c): add exactly ONE composed index to serve filter + aggregation.
--   Think about column order (equality cols first, range last) and whether
--   INCLUDE (value) lets you get an INDEX-ONLY scan.
--   CREATE INDEX IF NOT EXISTS ... ;  ANALYZE ghcn_raw;

-- TODO(d): re-run EXPLAIN (ANALYZE, BUFFERS) after (b) and again after (c).
--   Prove each change: operator switched, time dropped, buffers dropped.

-- TODO(e): in tuning.md, name ONE index you deliberately did NOT add and why.

• Improving Query Performance in PostgreSQL DataCamp · Course
• PostgreSQL Query Optimization, 2nd Ed. O'Reilly · Book

Profile the wide, denormalized NYC DOHMH inspection flat file in pandas to empirically discover functional dependencies and quantify redundancy, then design and build a 3NF schema in DuckDB (restaurant / inspection / violation + cuisine/borough lookups) with real PRIMARY KEY and FOREIGN KEY constraints. Split the flat file into the normalized tables idempotently, then PROVE integrity by watching DuckDB reject an FK-violating insert and a duplicate-PK insert, add one deliberately justified denormalization, and ship the whole thing as your Phase 1 capstone README.

1. LAND THE DATA: page the Socrata API at https://data.cityofnewyork.us/resource/43nn-pn8j.json using $limit/$offset (pull ~50k rows) into a raw DataFrame, and write it verbatim to a local raw_inspections.parquet so every rerun starts from the same immutable snapshot (no re-hitting the API).
2. PROFILE FOR FDs: in pandas, quantify the redundancy — e.g. groupby('camis') and confirm dba/building/street/boro/cuisine_description are (nearly) constant per CAMIS; count how many times each restaurant's name+address is physically repeated across rows; flag the CAMIS values where the 'constant' columns actually disagree (schema drift / late edits) as candidate dirty keys.
3. DISCOVER KEYS & ANOMALIES: verify candidate keys empirically — test whether camis alone, (camis, inspection_date, inspection_type), and (camis, inspection_date, inspection_type, violation_code) are unique; write down which grain each represents (restaurant vs inspection vs violation) and note the nulls in violation_code/grade/inspection_date that break naive keys.
4. MODEL THE ERD: draw the target 3NF schema as a Mermaid erDiagram (or dbdiagram.io) — restaurant (PK camis), inspection (PK inspection_id derived from camis+date+type), violation (PK/composite FK to inspection + violation_code), plus cuisine and boro lookup tables — showing every FK relationship and cardinality; commit the .mmd/source into the repo.
5. WRITE DDL: fill in the starter scaffold to CREATE the staging table plus the five target tables in DuckDB with explicit PRIMARY KEY and FOREIGN KEY constraints, choosing column types deliberately (dates as DATE, score as INTEGER) and documenting why each PK was chosen in a comment.
6. SPLIT IDEMPOTENTLY: load raw_inspections.parquet into staging, then run INSERT ... SELECT DISTINCT statements (lookups first, then restaurant, inspection, violation) to populate the normalized tables; make reruns safe by using CREATE OR REPLACE / DELETE-before-load / anti-join so a second run produces zero duplicate rows — state which idempotency strategy you chose and why.
7. QUARANTINE BAD ROWS: before the FK-constrained inserts, route rows that would violate integrity (null camis, unparseable inspection_date, violation_code not tied to a valid inspection) into a rejected_rows table with a reason column and a load timestamp instead of dropping them silently.
8. PROVE INTEGRITY: run two inserts that MUST fail and capture the DuckDB error text in your README — (a) an INSERT into violation referencing a camis/inspection_id that does not exist (FK violation), and (b) an INSERT into restaurant reusing an existing camis (duplicate PK); show that both are rejected and explain in one line what each proves.
9. JUSTIFY ONE DENORMALIZATION + SHIP README: add exactly one deliberate denormalization (e.g. carry cuisine_description onto inspection, or store latest_grade on restaurant) and write 2-3 sentences on the read/query win vs the update-anomaly risk you re-accepted; then assemble the Phase 1 portfolio README (business framing, ERD image, DDL, the two rejected inserts, the why-notes) so a hiring manager can clone and read it in 30 seconds.

• DuckDB — CREATE TABLE, PRIMARY KEY & FOREIGN KEY constraints
• DuckDB — Constraints (primary/foreign/unique/not null) reference
• Normalization: 1NF → 2NF → 3NF → BCNF explained
• Mermaid Entity-Relationship Diagram syntax

-- ============================================================
-- Week 13 — Normalize the NYC DOHMH flat file into 3NF (DuckDB)
-- Load raw_inspections.parquet (landed in the pandas step) first.
-- ============================================================

-- 0) STAGING: mirror the wide flat file exactly, no constraints.
CREATE OR REPLACE TABLE stg_inspections AS
SELECT * FROM read_parquet('raw_inspections.parquet');
-- TODO: sanity-check row count + peek at columns (camis, dba, boro,
--       cuisine_description, inspection_date, inspection_type,
--       violation_code, score, grade).

-- 1) LOOKUP TABLES (small dimensions the fact rows repeat endlessly)
CREATE OR REPLACE TABLE boro (
    boro_name TEXT PRIMARY KEY
);
CREATE OR REPLACE TABLE cuisine (
    cuisine_description TEXT PRIMARY KEY
);
-- TODO: INSERT ... SELECT DISTINCT the non-null values from staging.
--       Decide how to treat '' / '0' / NULL boro codes.

-- 2) RESTAURANT  (grain = one row per CAMIS)
CREATE OR REPLACE TABLE restaurant (
    camis               TEXT PRIMARY KEY,
    dba                 TEXT,
    building            TEXT,
    street              TEXT,
    zipcode             TEXT,
    boro_name           TEXT REFERENCES boro(boro_name),
    cuisine_description TEXT REFERENCES cuisine(cuisine_description)
    -- TODO (denormalization decision): should latest_grade live here?
    --      If you add it, justify the update-anomaly risk in the README.
);
-- TODO: INSERT ... SELECT — collapse many rows per camis to ONE.
--       A restaurant can have conflicting name/address strings across
--       rows: pick a rule (e.g. most recent inspection_date wins) and
--       write down why. Rows with NULL camis must NOT land here.

-- 3) INSPECTION (grain = one visit: camis + date + type)
CREATE OR REPLACE TABLE inspection (
    inspection_id   TEXT PRIMARY KEY,          -- TODO: derive a stable id
    camis           TEXT NOT NULL REFERENCES restaurant(camis),
    inspection_date DATE,
    inspection_type TEXT,
    score           INTEGER,
    grade           TEXT
    -- TODO: pick your ONE justified denormalization if it belongs here
    --       (e.g. carry cuisine_description for fast filtering).
);
-- TODO: build inspection_id (e.g. hash/concat of camis||date||type),
--       CAST inspection_date TEXT -> DATE, and INSERT SELECT DISTINCT.
--       Idempotency: use CREATE OR REPLACE or an anti-join so a rerun
--       adds zero duplicates — state your choice.

-- 4) VIOLATION (grain = one violation on one inspection)
CREATE OR REPLACE TABLE violation (
    inspection_id  TEXT NOT NULL REFERENCES inspection(inspection_id),
    violation_code TEXT NOT NULL,
    description    TEXT,
    critical_flag  TEXT,
    PRIMARY KEY (inspection_id, violation_code)   -- TODO: verify unique
);
-- TODO: INSERT SELECT — rows with NULL violation_code are not violations;
--       route them (and orphan inspection_ids) to rejected_rows instead.

-- 5) QUARANTINE
-- CREATE OR REPLACE TABLE rejected_rows (
--     reason        TEXT,
--     rejected_at   TIMESTAMP DEFAULT now(),
--     raw_camis     TEXT,
--     raw_row       TEXT
-- );
-- TODO: before the FK inserts, capture null-key / bad-date / orphan rows.

-- 6) PROVE INTEGRITY (these two statements MUST fail — paste errors in README)
-- INSERT INTO violation VALUES ('does-not-exist','10F','x','N');   -- FK reject
-- INSERT INTO restaurant (camis) VALUES ('<an-existing-camis>');   -- dup PK reject

• Data Warehouse Schemas Explained

• Database Design DataCamp · Course
• Database Design for Mere Mortals (25th Anniversary Ed.) O'Reilly · Book