# gharchive ETL — sample pipeline

This directory hosts the **canonical sample pipeline** built across the
*Data Engineering with Python* course. Every lesson edits some subset of
these files; by the end of the course you have a productionised lakehouse
pipeline with observability, lineage, and cost controls.

The pipeline processes the public [gharchive.org](https://www.gharchive.org/)
dataset — hourly JSON dumps of every public GitHub event since 2011. No
auth, no quotas, terabytes available — perfect for real-world ETL pain
(deduping, schema drift, time zones, late-arriving data).

## Lesson -> expected state map

| Lesson | Sub-tree touched | State at end of lesson |
|---|---|---|
| 01 | `extract/`, `transform/` | 50-line `extract/gharchive.py` downloads one hour; `transform/events.py` counts events by type; output is a CSV. |
| 02 | `extract/`, `transform/` | Same logic, but reading/writing **Parquet** via `pyarrow`; partitioned by event date. |
| 03 | `quality/` | First Great Expectations suite: schema validation, null checks, type checks; runs against the L02 Parquet output. |
| 04 | `airflow/` (added) | First DAG wraps the L01-L03 scripts; runs in docker-compose. |
| 05 | `airflow/` | Schedule intervals, catchup, manual backfills. |
| 06 | `airflow/`, `transform/` | Idempotent upserts; retry budgets; XCom hygiene. |
| 07 | `transform/` (SQL) | SQL transforms (window functions, CTEs) added alongside the pandas ones. |
| 08 | `dbt/` — `dbt_project.yml`, 2 staging models, 2 mart models, schema tests, 1 singular test, `profiles.yml.example` | Adds the dbt transformation layer on top of the Parquet output. Builds against DuckDB at `warehouse/dev.duckdb`. Run with `cd dbt && DBT_PROFILES_DIR=$(pwd) dbt build`. |
| 09 | `duckdb/` | DuckDB query layer over the L02 Parquet partitions. Three scripts: `query.py` (shared connection helper), `count_by_type.py`, `top_repos.py`. |
| 10 | `iceberg/` | Iceberg table layer. SqlCatalog at `warehouse/catalog.db`; the `default.events` table is populated by `write_from_parquet.py` (renames source `date` column to `event_date`) and queried by `query_duckdb.py` via DuckDB's iceberg extension. `maintenance.py` runs snapshot expiration + small-file rewrites. |
| 10 | `iceberg/` (added) | Iceberg tables with snapshots, time travel, ACID on the lake. |
| L11 | `warehouse/`  | Cloud-warehouse adapters (two-track). `snowflake_adapter.py` and `bigquery_adapter.py` register the L10 Iceberg table as an external table and materialise `daily_counts` (Track A — requires cloud env vars). `duckdb_local.py` is the Track B fallback that exposes the same surface against the local Iceberg. `materialize_vs_query.py` benchmarks external-table vs CTAS-materialised reads. |
| 12 | `observability/` (added) | OpenLineage emit helper (`openlineage_emit.py`) with `file://` fallback; freshness SLA checker (`freshness.py`, exit 0/1/2); stdlib alert helper (`alerts.py`); Marquez docker-compose (`marquez-docker-compose.yml`). DAG (`gharchive_hourly`) gains `pre_execute`/`post_execute` OpenLineage hooks on every task and a final `freshness_check` task that raises on breach. |
| 13 | `schema_evolution/` (added) | Pytest suite proving backward/forward compatibility across v1/v2/v3 Parquet fixtures; `data_contract.yml` declares producer/consumer schema guarantees and deprecation policy; `generate_fixtures.py` reproduces the golden fixtures. |
| 14 | `governance/` (added), `warehouse/adapter.py` (modified) | Cost attribution helpers (`cost_attribution.py`); declarative table-level ACLs (`grants.yml`); GDPR hard-delete with Iceberg snapshot expiration (`gdpr/right_to_be_forgotten.py`). Warehouse adapter gains a `tag` kwarg for query-cost attribution; DuckDB records elapsed seconds in `_query_costs`. |

## Layout (after Plan A — current)

```
sample-pipeline/
  README.md
  requirements.txt
  extract/
    __init__.py
    gharchive.py     # stub; real code in L01
  transform/
    __init__.py
    events.py        # stub; real code in L01 (CSV) -> L02 (Parquet)
  quality/
    __init__.py
    expectations.py  # stub; real code in L03
```

## Layout (full course — after Plan E)

```
sample-pipeline/
  extract/          # L01
  transform/        # L01-L08, L13
  quality/          # L03
  airflow/          # L04-L06
  dbt/              # L08
  iceberg/          # L10
  warehouse/        # L11
  observability/    # L12
  requirements.txt
  README.md
```

Plan A delivers only the *skeleton* (empty modules with docstrings).
Real code arrives lesson by lesson in Plans B–E.

## Quick start (after L01)

```bash
cd data-engineering/sample-pipeline
python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt
python -m extract.gharchive --hour 2024-01-01-15  # downloads one hour
python -m transform.events --input data/raw/2024-01-01-15.json.gz
```

(The commands above are aspirational targets for lesson 01 — the stubs
in this Plan-A skeleton only print "not implemented yet" messages.)