Iki-ETL-Table-Driven (ikietl)

A single, table-driven ETL Swiss-army-knife. Every stage — extract → transform → validate → load — is described as data (arrays of stage-specs) against any file location or any database, and a small facade hides the pipeline DAG underneath.

This is not just a config-file tool. The same pipeline definition can come from a YAML/TOML/JSON file or be built entirely in Python as a plain dict and handed to build_pipeline_config(...) — no file on disk required. The CLI (etl) is a thin wrapper over a Python package (ikietl) that you can import and drive directly: construct configs programmatically, inspect PipelineConfig objects, call run_pipeline()/validate()/status() from your own scripts, orchestrator, or test suite. See Python API below.

etl run | dry-run | validate | status | audit | clean | init

• Table-driven — the pipeline is arrays of declarative stage-specs, not code
• Config file or Python dict — write ikietl.yaml and use the CLI, or call build_pipeline_config({...}) from Python and skip files entirely
• Universal source — any DB (SQLAlchemy) + any storage backend (fsspec) + any file format (registry), read/written with either pandas or polars
• Declarative transforms — ops for the common cases (rename, filter, join, dedup, …), plus a restricted expression evaluator for row-level logic
• Schema-validated config — Pydantic catches structural mistakes before a byte is read, whether the config came from a file or a dict
• Examples-first — a gallery of runnable ETL scenarios, each with its own workspace under examples/

Before you build on this: an independent audit of this codebase found two advertised features that do not currently work (op: custom transforms, and SQL mode: upsert), plus a couple of path-resolution edge cases. Everything else described below has been verified by running the actual code.

---

Install

pip install ikietl

One package, no extras to remember for the default setup — it ships with everything the tool supports out of the box: cloud filesystems (s3fs, gcsfs, adlfs), Postgres/MySQL drivers (psycopg2-binary, pymysql), and Excel (openpyxl). All file I/O runs on pandas by default. (There is currently no bundled SQL Server driver — add pyodbc or pymssql yourself and pass the matching SQLAlchemy connection string if you need it.)

If you also want the polars I/O engine (see I/O engines: pandas & polars below), install the optional extra:

pip install "ikietl[polars]"

ikietl works exactly the same without it — polars is opt-in per extract/load step, and you only need it installed if a step actually requests engine: polars.

Editable/dev install:

pip install -e .
pip install pytest   # only needed to run the test suite

---

60-second quickstart

etl init                            # scaffold ikietl.yaml + work/logs/state/audit/tmp/transforms
etl validate                        # config schema + data-contract checks (extract -> transform -> validate, no load)
etl dry-run                         # same stages as validate
etl run                             # execute the full pipeline, including load
etl status                          # show last run's state (timestamp, dataset row counts)
etl audit                           # show the lineage log (one line per stage per dataset)
etl clean                           # remove work/ and tmp/ (state/ is kept)

etl looks for ikietl.yaml / ikietl.yml / ikietl.toml / ikietl.json in the current directory by default; pass --config/-c to point at a specific file. All three formats parse into the same validated Pydantic model, so every command below works identically regardless of which one you use.

Try the example gallery

The repository ships with several runnable example scenarios under examples/, each backed by a small Python script that seeds sample data, writes a matching ikietl.yaml, and runs the pipeline through the real engine (no network or database server required — SQL examples use an in-memory/temp SQLite database):

python -c "from examples.basic_examples import run_database_etl_example; print(run_database_etl_example())"
python -c "from examples.basic_examples import run_cloud_storage_etl_example; print(run_cloud_storage_etl_example())"

Every function in examples/basic_examples.py (run_database_etl_example, run_incremental_etl_example, run_mixed_source_etl_example, run_quality_monitoring_etl_example, run_event_stream_etl_example, run_unstructured_data_etl_example, run_cloud_storage_etl_example, and several smaller run_*_example helpers) is self-contained and safe to call directly — they're also exercised by tests/test_examples.py.

Each scenario also has a matching folder (e.g. examples/database-etl/) containing the ikietl.yaml it produces plus input/, output/, work/, logs/, state/, and tmp/ sub-folders — inspect those after running the example above to see exactly what the pipeline read and wrote.

Note: there are currently no Docker Compose files in this repository for Postgres/MySQL/MinIO backends — the "database" and "cloud-storage" examples run entirely against local SQLite/filesystem stand-ins so they work with no setup. If you want to point them at a real Postgres/MySQL/S3 endpoint, supply your own connection string / bucket via the conn / path fields and (for cloud storage) the relevant storage_options.

---

How a config file is structured

Every ikietl.yaml (or .toml/.json) has up to five top-level sections:

meta: { ... } # required — name/version/description
runtime: { ... } # optional — work/log/state/tmp dirs, strict mode, parallelism
extract: [...] # required — where data comes from
transform: [...] # optional — how it's reshaped
validate: [...] # optional — data-contract checks
load: [...] # required — where results go

Datasets flow through the pipeline by id: each extract entry produces a named dataset; each transform entry reads one named dataset (input:) and produces a new one (its own id:); validate and load entries reference any dataset by id. IDs must be unique across extract + transform outputs.

meta — required

meta:
  name: my-pipeline
  version: "1.0" # optional, defaults to "1.0"
  description: "..." # optional, defaults to ""

runtime — optional, but read the caveat below

runtime:
  work_dir: ./work # default: ./work
  log_dir: ./logs # default: ./logs
  state_dir: ./state # default: ./state
  tmp_dir: ./tmp # default: ./tmp
  strict: true # default: true — any validation failure aborts before load
  parallelism: 1 # default: 1 — >1 extracts run concurrently in a thread pool

⚠️ Caveat: when runtime: is present (even partially), every path in it is resolved relative to the config file's own directory, not your current working directory — this is what lets you run etl from anywhere and still get consistent output locations. If you omit runtime: entirely, the defaults above are used as literal, un-resolved relative paths instead, which will resolve against whatever directory the process happens to be launched from. Recommendation: always include a runtime: block, even if you're only overriding one field, to get predictable, config-relative paths.

parallelism > 1 only affects the extract stage, and only when there's more than one extract entry — each extractor runs in its own thread. Extract results are still merged deterministically by dataset id afterwards, so downstream transform/validate/load behavior is unaffected by ordering.

extract — required, one or more sources

extract:
  - id: sales_csv
    type: file
    path: "s3://my-bucket/incoming/sales_*.csv" # any fsspec URL, or a local path
    format: csv # optional — inferred from the extension if omitted
    options:
      delimiter: ","
      encoding: "utf-8"
      engine: pandas # optional — "pandas" (default) or "polars", see below
    glob: true # true = read + concat every file matching the path

  - id: customers_db
    type: sql
    conn: "postgresql://user:${env.DB_PASS}@host/db" # any SQLAlchemy URL
    query: "SELECT * FROM customers WHERE updated_at > :last_run"
    params: { last_run: "${state.last_run}" } # bound as real SQL params — safe from injection

• ${env.VAR} interpolates an environment variable. Missing variables raise a clear ConfigError at load time rather than failing deep inside a run.
• ${state.key} interpolates a value from the previous run's state file (state/last_run.json) — the standard way to do incremental/watermark extraction (e.g. "only rows newer than last time I ran").
• {date} anywhere in a string is replaced with today's date (YYYY-MM-DD) before anything else is resolved — handy for date-partitioned paths.
• glob: true requires the path to actually match at least one file, or extraction fails loudly rather than silently returning nothing.

Local file paths (for both extract and load) are resolved relative to the config file's directory automatically — a script running etl from any other working directory still reads/writes in the same place relative to ikietl.yaml.

transform — optional, applied in step order

transform:
  - id: clean_sales
    input: sales_csv
    steps:
      - op: rename
        map: { "Order ID": order_id, "Amount": amount }
      - op: cast
        column: amount
        type: decimal # int | float | string | datetime | decimal
      - op: filter
        expr: "amount > 0"
      - op: derive
        column: year
        expr: "year(order_date)"
      - op: select
        columns: [order_id, amount, year, customer_id]
      - op: join
        with: customers_db # another dataset id — must already exist
        on_column:
          customer_id # NOT `on:` — bare `on:` parses as the YAML 1.1
          # boolean key `True` in some YAML loaders, and
          # is only accepted here as a fallback alias
        type: left # left | right | inner | outer (passed to pandas.merge)
      - op: sort
        by: [order_date]
        ascending: true
      - op: dedup
        columns: [order_id]

filter and derive expressions run through simpleeval against the row's columns (plus the previous run's state, if any) — never raw eval() — so a config file cannot execute arbitrary code through these two ops. Whitelisted helper functions: year(), month(), day(), upper(), lower(), abs(), round().

Every transform op (rename, select, drop, cast, filter, derive, join, sort, dedup) runs on a pandas.DataFrame, regardless of which I/O engine: produced the dataset — see the next section for why.

op: custom — supported. The codebase supports declaring a custom Python transform (op: custom) that points at a local module.py and a function name. module/function may be supplied at the step top-level or inside args:. Relative module paths are resolved relative to the config file's directory. See src/ikietl/transform/main.py for the implementation details.

validate — optional, data-contract checks

validate:
  - id: sales_quality
    input: clean_sales
    rules:
      - { type: not_null, columns: [order_id, amount] }
      - { type: unique, columns: [order_id] }
      - { type: range, column: amount, min: 0.01, max: 1000000 }
      - { type: row_count, min: 1 }

In strict: true mode (the default), any rule failure raises immediately and aborts the run before load executes. In strict: false mode, a failure is recorded to the audit/lineage log ({"failed": true, "error": "..."}) and the pipeline continues to the next validation/load step instead of stopping.

load — required, one or more destinations

load:
  - id: archive_csv
    input: clean_sales
    type: file
    path: "./work/clean_sales.csv"
    format: csv
    options:
      delimiter: ";"
      engine: polars # optional — write with polars instead of pandas

  - id: warehouse
    input: clean_sales
    type: sql
    conn: "postgresql://user:${env.DB_PASS}@host/warehouse"
    table: fact_sales
    mode: append # append | replace  — see caveat below
    transaction: true # wrap the write in a DB transaction (default: true)

• File loads are atomic: data is written to path + .tmp and then renamed into place, so a crash mid-write never leaves a half-written file at the real path.
• SQL mode supports append, replace, and upsert (key-based merge) — when mode: upsert is used you must set key: [col1, col2] on the load spec. upsert is implemented for Postgres and MySQL/MariaDB via a staging table + INSERT ... ON CONFLICT / ON DUPLICATE KEY UPDATE. insert is accepted as an alias for append.

---

I/O engines: pandas & polars

File extract/load is a table-driven registry: one row per (format, engine) pair, shared by every extractor and loader — adding a format means adding a row, not a new code path. Two engines are supported, selected per extract/load step with options.engine:

options:
  engine: pandas # default — used automatically if you omit `engine:`
  # engine: polars   # opt-in — requires `pip install "ikietl[polars]"`

Why both, and how they interoperate: transform/validate always operate on a pandas.DataFrame (that's the pipeline's internal contract), so an extract step with engine: polars reads with polars — for its speed and broader native-format support — and the result is converted to pandas immediately afterwards via Arrow before it reaches your transform: steps. A load step with engine: polars does the reverse: your pandas dataset is converted right before writing, so the fast columnar polars writer is used. Mixing engines across steps in the same pipeline is fine — a file written by one engine can always be read back by the other.

Format matrix

pandas (engine: pandas, the default) — mirrors the pandas I/O guide:

Format	Extension(s) inferred	Read	Write
CSV	.csv	✅	✅
TSV	.tsv	✅	✅
Fixed-width text	.txt	✅	—
JSON	.json	✅	✅
JSON Lines	.jsonl, .ndjson	✅	✅
HTML	.html, .htm	✅	✅
XML	.xml	✅	✅
Excel	.xlsx, .xls	✅	✅ (.xlsx)
OpenDocument	.ods	✅	—
Parquet	.parquet, .pq	✅	✅
Feather	.feather, .arrow, .ipc	✅	✅
ORC	.orc	✅	✅
HDF5	.h5, .hdf5	✅ *	✅ *
Stata	.dta	✅	✅
SAS	.sas7bdat	✅	—
SPSS	.sav	✅ *	—
Pickle	.pkl, .pickle	✅	✅

Blank cells (—) match pandas' own IO table — those formats genuinely have no writer upstream, not a gap in this registry.

polars (engine: polars, opt-in) — mirrors the polars I/O reference:

Format	Read	Write
CSV	✅	✅
TSV	✅	✅
JSON	✅	✅
JSON Lines (ndjson)	✅	✅
Parquet	✅	✅
Feather / IPC / Arrow	✅	✅
Excel	✅	✅
Avro	✅	✅

If a step requests a format the chosen engine doesn't support, extraction/load fails fast with a clear ValueError naming the format, the engine, and every format that engine does support — not a confusing library-internal error.

* HDF5 and SPSS require a real filesystem path, not an in-memory stream — that's a limitation of the underlying pytables/pyreadstat libraries, not this tool. Reading either format from a local path: works exactly like any other format. Reading from a remote path: (s3://, gs://, …) is still supported: the file is transparently staged to a local temp file first, then read, then the temp file is removed. Writing HDF5 to a remote path: is not supported — write it locally and upload separately.

Any reader/writer keyword arguments not covered by the first-class options fields above (delimiter, encoding, sheet, key, …) can be passed straight through via options.read_kwargs / options.write_kwargs, e.g.:

options:
  engine: pandas
  read_kwargs: { dtype: { order_id: str } }

---

Python API

Everything the CLI does is a thin wrapper around ikietl's public API — you can call the same functions directly from Python:

from ikietl import (
    audit, build_pipeline_config, clean, init,
    load_config, run_pipeline, status, validate,
)

Function	Signature	What it does
load_config(path)	(str | Path) -> PipelineConfig	Parse + validate a YAML/TOML/JSON file into a PipelineConfig.
build_pipeline_config(config_data, state_dir="./state")	(dict) -> PipelineConfig	Validate a plain Python dict into a PipelineConfig — no file needed at all.
run_pipeline(config, dry_run=False)	(PipelineConfig | str) -> dict	Run extract → transform → validate → (load unless dry_run). Accepts a PipelineConfig object or a path string. Returns {"datasets": {id: row_count}, "state_dir": ..., "audit_dir": ...}.
validate(config)	(PipelineConfig | str) -> dict	Same as run_pipeline(config, dry_run=True).
status(config)	(PipelineConfig | str) -> dict	Read state/last_run.json for this config.
audit(config)	(PipelineConfig | str) -> str	Read the raw logs/audit/lineage.jsonl contents.
clean(config)	(PipelineConfig | str) -> None	Delete work//tmp/ contents (keeps state/).
init(name="my-pipeline")	(str) -> None	Scaffold a starter ikietl.yaml + directories in the current directory.

Fully in-memory — no config file on disk

Because build_pipeline_config takes a plain dict, you can define, run, and inspect a whole pipeline from a script or notebook without ever writing an ikietl.yaml:

from ikietl import build_pipeline_config, run_pipeline

cfg = build_pipeline_config({
    "meta": {"name": "in-memory-demo", "version": "1.0"},
    "runtime": {
        "work_dir": "./work", "log_dir": "./logs",
        "state_dir": "./state", "tmp_dir": "./tmp",
        "strict": True,
    },
    "extract": [{
        "id": "sales_csv", "type": "file",
        "path": "./input/sample_sales.csv", "format": "csv",
    }],
    "transform": [{
        "id": "clean_sales", "input": "sales_csv",
        "steps": [
            {"op": "cast", "column": "amount", "type": "decimal"},
            {"op": "filter", "expr": "amount > 0"},
        ],
    }],
    "validate": [{
        "id": "sales_quality", "input": "clean_sales",
        "rules": [{"type": "row_count", "min": 1}],
    }],
    "load": [{
        "id": "archive", "input": "clean_sales",
        "type": "file", "path": "./output/output.csv", "format": "csv",
    }],
})

result = run_pipeline(cfg)         # dry_run=True to skip the load stage
print(result["datasets"])          # {"clean_sales": 3}

This is exactly the pattern used throughout examples/basic_examples.py (run_extractors_example, run_pipeline_config_example, etc.) — it's a first-class, tested way to use the package, not a fallback.

Calling the format registry directly

Outside of a full pipeline, the same table-driven registry that backs extract/load is importable and usable on its own — handy in a notebook or a one-off script:

from ikietl.formats.registry import read, write, infer_format, available_formats

available_formats("polars")   # {"csv": True, "json": True, "parquet": True, ...}
infer_format("s3://bucket/events.jsonl")   # "jsonl"

with open("sales.parquet", "rb") as f:
    df = read("parquet", f, options={}, engine="pandas")

write(df, "csv", "out.csv", options={"delimiter": ";"}, engine="pandas")

Loading a file, then tweaking it in Python before running

load_config() returns a real PipelineConfig (a Pydantic model), so you can mutate fields — e.g. redirect runtime directories for a test run — before passing it to run_pipeline:

from ikietl import load_config, validate, run_pipeline, status, audit, clean

cfg = load_config("ikietl.yaml")
cfg.runtime.work_dir = "/tmp/my-run/work"
cfg.runtime.state_dir = "/tmp/my-run/state"

validate(cfg)                # raises PipelineError if schema/data checks fail
run_pipeline(cfg)            # full run, load included
print(status(cfg))           # {"timestamp": ..., "datasets": {...}}
print(audit(cfg))            # raw lineage.jsonl contents
clean(cfg)                   # tidy up work/tmp when you're done

run_pipeline/validate/status/audit/clean all accept either a PipelineConfig object or a path string — pick whichever is convenient at each call site; both forms go through the identical schema validation and engine code the CLI uses, so behavior never diverges between "config file" and "pure Python" usage.

---

Reference: every extractor / transform op / validate rule / load mode

Stage	Options	Notes
Extract → file	path, format, options (incl. engine), glob	Any fsspec location: local, s3://, gs://, abfs://, http(s)://, sftp://. glob: true concatenates every matched file. options.engine picks pandas (default) or polars.
Extract → sql	conn, query, params	Any SQLAlchemy-supported database. params are bound as real query parameters (not string-interpolated), so this is not vulnerable to SQL injection from ${state.*} values.
Formats (read + write)	17 pandas formats + 8 polars formats — see I/O engines	Table-driven registry, one row per (format, engine), shared by every extractor/loader. format: is optional and inferred from the file extension when omitted.
Transform ops	rename, select, drop, cast, filter, derive, join, sort, dedup	filter/derive run through a restricted evaluator — no arbitrary code exec. custom is present but currently broken — see caveat above.
Validate rules	not_null, unique, range, row_count	In strict mode any failure aborts before load; in non-strict mode it's logged and the run continues.
Load → file	path, format, options (incl. engine)	Atomic write-then-rename. {date} in path is replaced with today's date. options.engine picks pandas (default) or polars.
Load → sql	conn, table, mode, transaction	mode: append or replace only — no upsert, and the key: field is currently unused.

---

CLI reference

Command	What it does
etl init [name]	Scaffold a new ikietl.yaml plus work/ logs/ state/ tmp/
etl validate	Config schema and data-contract checks — runs extract → transform → validate, no load
etl dry-run	Same stages as validate; framed as "no load executed"
etl run	The full pipeline, including load
etl status	Prints the contents of state/last_run.json (timestamp + per-dataset row counts)
etl audit	Prints logs/audit/lineage.jsonl — one JSON line per stage per dataset
etl clean	Deletes work/ and tmp/; keeps state/

All commands accept --config/-c <path>. Every one of them is a thin wrapper over the Python API — see Python API above for the full picture, including running pipelines with no config file at all.

---

Error-proofing, at a glance

Feature	How
Config schema validation	Pydantic model catches structural errors before run
Fail-fast / strict mode	Any validation rule failure aborts the run (configurable)
Transactional SQL loads	SQL loads wrapped in a transaction by default (transaction: true)
Atomic file loads	Written to path.tmp then renamed — never a half-written file
Format/engine mismatches	Clear ValueError naming the format, engine, and every format that engine supports — not a bare library error
State tracking	state/last_run.json — timestamp of each run + row counts per dataset
Audit / lineage	Each extract/transform stage appends {timestamp, stage, dataset_id, ...} to logs/audit/lineage.jsonl
Secrets	${env.VAR} interpolation — credentials never hardcoded in config
Exit codes	0 success · 1 validation · 2 extract · 3 transform · 4 load · 5 config

Sample etl audit output (actual field names — timestamp/dataset_id, not ts/dataset):

{"timestamp": "2026-07-26T10:00:00+00:00", "stage": "extract", "dataset_id": "sales_csv", "rows": 1000, "type": "file"}
{"timestamp": "2026-07-26T10:00:01+00:00", "stage": "transform", "dataset_id": "clean_sales", "input": "sales_csv", "rows": 950}

Note: only extract and transform stages are logged on success today; a failed non-strict validate rule is logged with "failed": true, and successful validations and every load currently produce no lineage entry.

---

Project layout

project/
├── ikietl.yaml           # your config (yaml/toml/json — pick one)
├── work/                 # intermediate/output artifacts written by `etl run`
├── state/                # last_run.json (timestamp + row counts)
├── logs/
│   └── audit/            # lineage.jsonl
├── tmp/                  # scratch, cleared by `etl clean`
└── examples/             # runnable ETL scenarios and per-example workspaces
    ├── basic-example/
    ├── database-etl/
    ├── cloud-storage-etl/
    └── ...

etl init scaffolds work/, logs/, state/, and tmp/ for you alongside a starter ikietl.yaml.

---

Testing

pip install -e .
pip install pytest
pytest

tests/test_pipeline_end_to_end.py builds a temporary config + CSV on the fly and runs it through the real pipeline (run_pipeline, the CLI's validate command, and the Python API's validate/status), asserting on row counts, output file contents, and config-relative path resolution.

tests/test_examples.py runs the shipped example scenarios end to end.

Known gap: the current test suite does not exercise op: custom, mode: upsert, runtime:-omitted configs, parallelism > 1, or the engine: polars I/O path — which is why some of these issues went unnoticed. If you add tests for these, please contribute them back.

---

License

MIT