ETL Pipeline

A three-stage pipeline that stays out of your way: extract CSV rows, transform them with plain functions, write the result. Every stage is a list-of-dicts in, list-of-dicts out, so you can insert your own stages without inheriting from anything or learning a DSL.

Package Layout

scripts/
  pipeline/
    __init__.py      # run_pipeline composition entry
    io_ops.py        # CSV reading / compact writing
    transform.py     # record-level transform stages
    sinks.py         # alternative output sinks (JSON)

When to Use

• Batch-normalizing a CSV export before loading it elsewhere.
• Teaching examples: a pipeline small enough to read in one sitting.
• Agent workflows that need deterministic, repeatable data passes.

Usage

import sys, os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "scripts"))

from pipeline import run_pipeline

run_pipeline("input.csv", "output.parquet")

Using the stages directly:

from pipeline.io_ops import read_csv
from pipeline.transform import transform

rows = read_csv("input.csv")
staged = transform(rows)          # your own steps here

Writing JSON instead:

from pipeline.sinks import write_json
write_json(staged, "out.json")

Notes

• run_pipeline composes read -> transform -> write; swap any stage freely.
• The transform stage is intentionally trivial: it marks rows as normalized. Real projects replace it with their business logic.