ETL (Extract, Transform, Load)

Introduction: Pipelines, Not One-Off Scripts

Extract pulls data from sources (APIs, DBs, files). Transform applies cleaning, joins, labeling merges, and business rules. Load writes curated datasets to the warehouse, object store, or feature store. In ML systems, transform often includes train/val/test splits and leakage checks from Data Leakage.

Reference Architecture

Batch vs Streaming

Most model training still uses batch curated tables; streaming feeds online features and monitoring. Volume 03 async I/O helps high-concurrency extract from APIs inside batch workers.

A Minimal Python ETL Job

"""ETL: raw events + labels -> curated training table"""
from pathlib import Path
import logging
import pandas as pd

logging.basicConfig(level=logging.INFO)
log = logging.getLogger("etl.train_table")

RAW_EVENTS = Path("data/raw/events")
RAW_LABELS = Path("data/labels")
CURATED = Path("data/curated")
CURATED.mkdir(parents=True, exist_ok=True)

def extract(batch_date: str) -> tuple[pd.DataFrame, pd.DataFrame]:
    events = pd.read_parquet(RAW_EVENTS / f"dt={batch_date}")
    labels = pd.read_csv(RAW_LABELS / f"labels_{batch_date}.csv")
    log.info("extracted events=%s labels=%s", len(events), len(labels))
    return events, labels

def transform(events: pd.DataFrame, labels: pd.DataFrame) -> pd.DataFrame:
    events = events.drop_duplicates(subset=["event_id"], keep="last")
    events["event_ts"] = pd.to_datetime(events["event_ts"], utc=True)
    df = events.merge(labels[["event_id", "label", "guideline_version"]], on="event_id", how="inner")
    df = df.loc[df["label"].notna()]
    return df

def load(df: pd.DataFrame, batch_date: str) -> Path:
    out = CURATED / f"train_table/dt={batch_date}" / "data.parquet"
    out.parent.mkdir(parents=True, exist_ok=True)
    df.to_parquet(out, index=False)
    log.info("loaded rows=%s path=%s", len(df), out)
    return out

def run(batch_date: str) -> Path:
    events, labels = extract(batch_date)
    curated = transform(events, labels)
    return load(curated, batch_date)

if __name__ == "__main__":
    run("2026-07-13")

Idempotency and Partitions

Re-running the same batch date should produce the same output path without duplicating rows in downstream consumers. Partition by dt=YYYY-MM-DD or batch_id so backfills replace one folder, not the entire dataset.

# Idempotent load — overwrite partition directory
partition = CURATED / "train_table" / "dt=2026-07-13"
if partition.exists():
    for f in partition.glob("*.parquet"):
        f.unlink()
df.to_parquet(partition / "data.parquet", index=False)

Quality Gates in Load

Knowledge Check