Skip to content

Getting Started

ETLRelay 0.7.0 supports an installed command-line interface (CLI), versioned YAML configuration, and direct Python composition. This guide starts with installing the CLI and running a YAML pipeline, then explains the configuration format and direct component API.

Requirements

  • Python 3.12 or newer.
  • ETLRelay 0.7.0.
  • A local filesystem directory where the example data and outputs can be stored.

Install the CLI

Install the released package into the Python environment where you want to run pipelines:

python -m pip install etlrelay==0.7.0
etlrelay --help
etlrelay version

For the full runnable examples, clone the repository and install the package in an environment using the repository's supported Python setup:

git clone https://github.com/apaquette/ETLRelay.git
cd ETLRelay
python -m pip install -e .

The examples are kept under examples/configuration/ so that their YAML files, preparation script, and runner can be used together. If you are using the published package rather than a checkout, obtain the examples from the v0.7.0 repository tag.

Prepare the demonstration data

Run the preparation script from the repository root:

python examples/configuration/prepare_demo_data.py

It creates a small CSV input file, a corresponding Parquet input file, and source/target SQLite databases in examples/configuration/data/. The target table is deliberately created before the SQL sink runs because SqlSink appends to an existing table; it does not create or migrate tables.

The sample input contains id, amount, and state. The YAML processors select id and amount, keep rows where amount >= 0, and rename id to record_id.

Run a YAML pipeline

After preparing the demonstration data, run commands from the repository root. The CLI loads and validates the YAML configuration, builds the pipeline with ETLRelay's existing component catalog and PipelineBuilder, executes the pipeline, and releases builder-managed resources.

Validate and execute a CSV-to-CSV pipeline:

etlrelay validate examples/configuration/csv_to_csv.yaml
etlrelay run examples/configuration/csv_to_csv.yaml

A successful validation does not execute the pipeline or create the sink output. Run the run command to perform the data transformation.

Other examples can be run in the same way. For example:

etlrelay validate examples/configuration/csv_to_parquet.yaml
etlrelay run examples/configuration/csv_to_parquet.yaml

The repository contains nine configuration examples under examples/configuration/: file-to-file format pairs, file-to-SQL, SQL-to-file, and SQL-to-SQL. Outputs are written beneath examples/configuration/data/ according to each configuration's storage settings.

For SQL examples, configure the demo database URLs in the process environment:

export ETLRELAY_SOURCE_DATABASE_URL="sqlite:///examples/configuration/data/source.db"
export ETLRELAY_SINK_DATABASE_URL="sqlite:///examples/configuration/data/sink.db"

Then validate and run a SQL pipeline, for example:

etlrelay validate examples/configuration/sql_to_sql.yaml
etlrelay run examples/configuration/sql_to_sql.yaml

Each SQL configuration references the variable name in its connection_url_env field. The YAML file does not contain the resolved URL. These demo URLs are local and contain no credentials; do not put credential-bearing connection URLs or passwords into configuration files. The destination table must exist before the SQL sink runs; the preparation script creates the demonstration tables.

See the CLI Reference for all commands, exit codes, and error-output conventions.

Understand the YAML pipeline

A typical file pipeline has four required top-level properties: version, source, processors, and sink.

version: 1

source:
  type: file
  path: examples/configuration/data/input.csv
  format: csv
  storage:
    type: local
    base_path: .

processors:
  - type: select_columns
    columns: [id, amount]
  - type: filter_rows
    column: amount
    operator: ">="
    value: 0
  - type: rename_columns
    mapping:
      id: record_id

sink:
  type: file
  path: examples/configuration/data/output.csv
  format: csv
  storage:
    type: local
    base_path: .

The version property is the configuration schema version, currently integer 1; it is independent of the Python package version 0.7.0. A processors list is required, but processors: [] is valid and passes batches through without transformations.

Processor order matters: select_columns drops state, filter_rows removes negative amounts, and rename_columns changes the retained id field to record_id. Formats are explicit; ETLRelay does not infer them from file extensions.

Validation and errors

The loader's validation stage checks the configuration structure, required fields, schema version, component names, and property types. Invalid configuration is rejected before source/sink operations and pipeline execution.

For example:

  • An unknown property such as sink.destination is rejected.
  • An unsupported file format is rejected using the format registry's reading/writing capability.
  • A filter operator outside ==, !=, <, <=, >, >= is rejected.
  • A missing or empty SQL connection environment variable fails during build-time preflight, before ETL component factories are invoked.

Schema validation does not prove that the named input file exists, that a selected column is present in the runtime schema, that SQL will succeed, or that the destination table is compatible. These are runtime conditions. See YAML Configuration for the full schema and Configuration Validation Architecture for validation boundaries.

Programmatic composition remains supported

The YAML interface is additive; direct Python construction remains available. This is still supported:

from pathlib import Path

from etlrelay.pipeline import Pipeline
from etlrelay.processors import FilterRows, RenameColumns, SelectColumns
from etlrelay.readers import CsvReader
from etlrelay.security import PathValidator
from etlrelay.sinks import FileSink
from etlrelay.sources import FileSource
from etlrelay.storage import LocalFileSystem
from etlrelay.writers import CsvWriter

base = Path("data").resolve()
storage = LocalFileSystem(path_validator=PathValidator(base_path=base))
pipeline = Pipeline(
    source=FileSource(path="input.csv", storage=storage, reader=CsvReader()),
    processors=[
        SelectColumns(columns=["name", "age"]),
        RenameColumns(mapping={"name": "customer_name"}),
        FilterRows(column="age", operator=">=", value=18),
    ],
    sink=FileSink(path="output.csv", storage=storage, writer=CsvWriter()),
)
pipeline.run()

For this direct API example, data/input.csv must exist and the output is data/output.csv. Engine lifecycle for direct SqlSource/SqlSink components remains caller-owned; the managed engine lifecycle described above applies to configuration-built pipelines using the shared SqlEngineFactory.

More details