AskAiml
Sign in Get started

ETL pipeline with polars

CSV → Postgres with polars, asyncpg, retries, structured logging, and a CI-driven release pipeline.

🛠 Advanced

Goal

Build a reusable, production-grade ETL framework that ingests messy CSV / JSON / Parquet files, validates with Pydantic, transforms with Polars, and loads into Postgres with proper idempotency, retries, monitoring, and CLI ergonomics.

Why this project

Data engineering is a high-paying niche. This project demonstrates: schema enforcement, performance (Polars), reliability (retries, idempotency), observability (metrics, lineage), and operational maturity (config, scheduling, alerts).

User experience

powershell
# Single-shot run
flowetl run pipelines/orders.yml --input data/orders_2026_06_07.csv

# Schedule (uses APScheduler internally)
flowetl schedule pipelines/orders.yml --cron "0 */6 * * *"

# Backfill
flowetl backfill pipelines/orders.yml --from 2026-01-01 --to 2026-06-01

# Validate config / dry-run
flowetl validate pipelines/orders.yml
flowetl run pipelines/orders.yml --dry-run

Tech stack

  • Polars (lazy DataFrames, expressions).
  • Pydantic v2 (schema validation, settings).
  • asyncpg + SQLAlchemy Core (Postgres).
  • tenacity (retries).
  • structlog + prometheus_client (observability).
  • APScheduler (cron-style scheduling) or Prefect for stretch.
  • typer + rich (CLI).
  • DuckDB for ad-hoc analytics / staging area (optional).
  • pytest + hypothesis + testcontainers.

Architecture

src/flowetl/
├── __init__.py
├── cli.py
├── config.py                  # PipelineConfig (pydantic)
├── pipeline.py                # Orchestrates Extract -> Transform -> Load
├── extract/
│   ├── csv.py
│   ├── json_lines.py
│   ├── parquet.py
│   └── s3.py
├── transform/
│   ├── ops.py                 # dedupe, normalize, derive cols
│   └── expressions.py         # reusable Polars expr factories
├── load/
│   ├── postgres.py            # batched insert + ON CONFLICT
│   ├── duckdb.py
│   └── parquet.py
├── validation/
│   └── schema.py              # Pydantic models per pipeline
├── state.py                   # pipeline_runs table, idempotency
├── observability.py           # metrics, log_event
├── retry.py                   # tenacity policies
└── scheduling.py
pipelines/
└── orders.yml
tests/
├── conftest.py
├── test_extract.py
├── test_transform.py
├── test_load.py
└── test_end_to_end.py

Spec

Pipeline config (orders.yml)

yaml
name: orders_etl
schema_version: 1

source:
  type: csv
  path: "data/orders_{ds}.csv"     # {ds} = run date
  options:
    encoding: utf-8
    sep: ","
    has_header: true

schema:
  order_id: int
  customer_id: int
  amount: float
  currency: str
  status: str
  placed_at: datetime

transforms:
  - drop_nulls: [order_id, customer_id, amount]
  - filter: "status != 'cancelled'"
  - derive:
      total_usd: "amount * fx_rate(currency, 'USD')"
      day: "placed_at.dt.date"
  - dedupe: [order_id]

sink:
  type: postgres
  table: public.orders
  primary_key: [order_id]
  upsert: true
  batch_size: 5000

quality_checks:
  - assert: "amount > 0"
  - assert: "order_id is not null"
  - row_count: "> 0"

Extractor interface

python
from typing import Protocol
import polars as pl

class Extractor(Protocol):
    def extract(self, source_cfg: dict) -> pl.LazyFrame: ...

Pipeline run

  1. Parse config; resolve {ds} placeholders.
  2. Extract → LazyFrame.
  3. Validate schema (Pydantic on sampled rows).
  4. Apply transforms (compose Polars expressions).
  5. Run quality checks → fail fast.
  6. Sink (batched upsert).
  7. Write pipeline_runs row with status, rows in/out, duration, error.

Idempotency

  • Each run keyed by (pipeline_name, run_ts, source_path_hash).
  • Re-running same key: skip if already SUCCESS, retry if FAILED.
  • Use ON CONFLICT (order_id) DO UPDATE for upsert semantics.

Retries

  • Network/DB errors: exponential backoff 3 attempts.
  • Validation errors: don't retry, fail loudly.

Metrics

  • flowetl_rows_extracted_total{pipeline=...}
  • flowetl_rows_loaded_total{pipeline=...}
  • flowetl_run_duration_seconds_bucket
  • flowetl_errors_total{pipeline,stage,reason}

Acceptance criteria

  1. End-to-end: 1M-row CSV → Postgres in < 30 s on laptop.
  2. Memory: < 1 GB even for 10M-row inputs (Polars lazy + streaming).
  3. Idempotent: re-running same input doesn't duplicate or change rows.
  4. Quality check failure → exit code 2 with clear log.
  5. Test coverage > 85 %.
  6. mypy --strict.
  7. Docker image runs the pipeline given a mounted config.

Stretch goals

  • DAG mode: multi-pipeline DAGs (Prefect / custom). Topological sort + parallel execution.
  • Data lineage: emit OpenLineage events to Marquez/DataHub.
  • Schema drift detection: compare new schema vs last, alert.
  • Slack/email alerts on failure.
  • Web UI to view runs, metrics, latest data sample (FastAPI + HTMX).
  • S3 source/sink with multipart uploads.
  • Spark / Dask backend swap for cluster-scale data.
  • dbt integration: trigger dbt runs after load.
  • Streaming mode: tail Kafka topic + micro-batch upsert.

Key implementation hints

Lazy extract + transform

python
import polars as pl

def extract_csv(path: Path) -> pl.LazyFrame:
    return pl.scan_csv(path, try_parse_dates=True)

def apply_transforms(lf: pl.LazyFrame, ops: list[dict]) -> pl.LazyFrame:
    for op in ops:
        if "filter" in op:
            lf = lf.filter(pl.sql_expr(op["filter"]))
        elif "derive" in op:
            for col, expr in op["derive"].items():
                lf = lf.with_columns(pl.sql_expr(expr).alias(col))
        elif "dedupe" in op:
            lf = lf.unique(subset=op["dedupe"])
        elif "drop_nulls" in op:
            lf = lf.drop_nulls(subset=op["drop_nulls"])
    return lf

Batched upsert

python
async def load_postgres(df: pl.DataFrame, *, table: str, pk: list[str], batch_size: int):
    cols = df.columns
    insert_sql = f"""
        INSERT INTO {table} ({", ".join(cols)})
        VALUES ({", ".join(f"${i+1}" for i in range(len(cols)))})
        ON CONFLICT ({", ".join(pk)}) DO UPDATE SET
        {", ".join(f"{c} = EXCLUDED.{c}" for c in cols if c not in pk)}
    """
    async with pool.acquire() as conn:
        async with conn.transaction():
            for chunk in df.iter_slices(batch_size):
                await conn.executemany(insert_sql, chunk.rows())

Run state machine

python
class RunState(str, Enum):
    PENDING = "pending"
    RUNNING = "running"
    SUCCESS = "success"
    FAILED = "failed"
    SKIPPED = "skipped"

@dataclass
class PipelineRun:
    id: UUID
    pipeline_name: str
    run_ts: datetime
    source_path: str
    state: RunState
    rows_in: int = 0
    rows_out: int = 0
    error: str | None = None
    started_at: datetime | None = None
    finished_at: datetime | None = None

Quality assertions

python
def assert_check(df: pl.DataFrame, expr: str) -> None:
    failures = df.filter(~pl.sql_expr(expr)).height
    if failures:
        raise QualityCheckError(f"{failures} rows failed: {expr}")

Deliverables

  • GitHub repo with example pipelines (3+) and demo data.
  • Grafana dashboard JSON for Prometheus metrics.
  • 1-pager: "How flowetl handles 100M-row daily backfills."
  • (Optional) Blog post benchmarking against Pandas + SQLAlchemy.

Lessons exercised

  • 04_asyncio (DB pool, concurrency)
  • 05_data (Polars, databases — heavy)
  • 06_testing (testcontainers!)
  • 07_performance (Polars streaming, profiling)
  • 09_packaging (Docker, entry points)

Time estimate: 30–50 hours core; 80–120 hours with stretch + DAG mode + UI.