dataloader-util (0.3.7)
Installation
pip install --index-url dataloader-utilAbout this package
Config-driven CLI for loading tabular data from sources (local, S3) in formats (CSV, JSONL, Parquet) into targets (Postgres, Trino/Iceberg, Snowflake).
dataloader-util
dataloader-util is a config-driven Python CLI for moving tabular data from files or S3 into local files and analytical databases. It uses DuckDB as the in-process execution engine, so jobs can read common data formats, run SQL-style transformations, infer schemas, and write to targets from a single YAML or JSON config file.
What it does
A load job follows one pipeline:
source file(s) -> DuckDB view -> optional transforms -> target writer
A config file declares named pools of sources, targets,
transform groups, and standalone actions. The load command picks
one source, one target, and zero or more transform groups by name at
run time:
sources: { s3_csv: ..., local_parquet: ... }
targets: { warehouse: ..., archive: ... }
transforms: { clean: [...steps], camel: [...steps] }
actions: { repair_table: ..., sample_rows: ... }
# Load CLI:
dataloader-util load -c job.yaml \
--source s3_csv --target warehouse \
--transform clean --transform camel
# Action CLI:
dataloader-util run -c job.yaml --action sample_rows
This lets a single config file describe many pipelines (different sources, destinations, or transform chains) and operational actions (query checks, metadata repairs, ad-hoc parameterized SQL) and pick one per run without rewriting the file.
Supported capabilities:
- Sources: local filesystem paths/globs and S3 or S3-compatible object stores.
- Input formats: CSV, JSONL/NDJSON, and Parquet.
- Targets: local/S3 Parquet or CSV files, Postgres, Trino, Trino-managed Iceberg tables, and Snowflake.
- Transforms: free-form DuckDB SQL,
WHEREfilters, column casts, and column renames. - Actions: standalone parameterized Trino and Snowflake query execution via
dataloader-util run. - Write modes:
replace,append, and key-basedmergefor database targets. - Config validation: strict Pydantic validation catches unknown keys and invalid combinations before a job runs.
- Lenient missing inputs: absent source files or empty globs skip successfully by default, with
fail_on_missing: trueavailable for strict pipelines. - Secrets via environment:
${VAR}placeholders are resolved from environment variables at runtime.
Requirements
- Python
>=3.12 uvfor local development commands- Docker only if you want to run local integration services/tests
- Target-specific connectivity for Postgres, Trino, Snowflake, or S3
Installation and setup
From a checkout of this repository:
uv sync
uv run dataloader-util --help
uv run dataloader-util --version
The package exposes the console command:
dataloader-util
During development, prefer running it through uv run so the repository virtual environment is used.
Python SDK
Install the same release artifact into another Python application or an Airflow worker:
pip install --extra-index-url https://<forgejo>/api/packages/<owner>/pypi/simple dataloader-util
Run a load directly from Python:
from dataloader_util.sdk import DataloaderError, run_job
try:
result = run_job(
"job.yaml",
source_name="events",
target_name="warehouse",
transform_names=["clean"],
source_overrides={"path": "s3://bucket/day=2026-01-01/*.parquet"},
)
except DataloaderError:
# Log, retry, or let the application fail.
raise
print(result.write_result.rows, result.skipped)
run_job accepts a YAML/JSON path or a validated PipelineConfig. Runtime overrides are nested Python mappings and are applied before environment substitution and validation. run_action provides the same interface for entries under actions::
from dataloader_util.sdk import run_action
result = run_action(
"job.yaml",
action_name="sample_rows",
action_overrides={"params": ["AMD"]},
)
The SDK returns structured RunResult, WriteResult, and ActionResult objects and raises the existing DataloaderError hierarchy. When run_job creates its own DuckDB engine, RunResult.final_view is informational because that engine is closed before return. Pass an Engine explicitly only when the caller needs to keep that view alive.
Airflow 3 TaskFlow example
No Airflow provider or custom operator is required. Install dataloader-util in the worker image and call it from an ordinary task:
from airflow.sdk import dag, task
@dag(schedule="@daily", catchup=False)
def daily_events():
@task
def load_events(day: str) -> dict[str, int | bool]:
# Task-local import avoids loading database drivers during DAG parsing.
from dataloader_util.sdk import run_job
result = run_job(
"/opt/airflow/dags/config/events.yaml",
source_name="events",
target_name="warehouse",
transform_names=["clean"],
source_overrides={"path": f"s3://bucket/events/day={day}/*.parquet"},
)
# Keep XCom small and serialization-independent.
return {"rows": result.write_result.rows, "skipped": result.skipped}
load_events("{{ ds }}")
daily_events()
Keep credentials in Airflow connections, a secrets backend, or worker environment variables. Do not put secrets or full SDK result objects into XCom. Airflow remains responsible for retries, timeouts, scheduling, and concurrency.
Quickstart: local CSV to local Parquet
Create a small input file:
mkdir -p data output
cat > data/events.csv <<'CSV'
id,event_ts,user_id,amount
1,2026-01-01T00:00:00Z,u1,12.50
2,2026-01-02T00:00:00Z,u2,-3.00
3,2026-01-03T00:00:00Z,u3,7.25
CSV
Run the included example job:
uv run dataloader-util load --config examples/job_local_to_local.yaml
With one source and one target defined, the CLI picks them
implicitly. To pick a specific source/target by name, use
--source / --target:
uv run dataloader-util load --config examples/job_multi_source_target.yaml \
--source events_csv --target events_parquet \
--transform keep_positive
Validate without running:
uv run dataloader-util validate --config examples/job_local_to_local.yaml
Perform a dry run that reads the source, applies transforms, and prints the planned target information without writing:
uv run dataloader-util load --dry-run --config examples/job_local_to_local.yaml
Enable debug logging:
uv run dataloader-util load --log-level DEBUG --config examples/job_local_to_local.yaml
CLI reference
dataloader-util [OPTIONS] COMMAND [ARGS]...
Global options:
| Option | Description |
|---|---|
--version |
Print the installed version and exit. |
--help |
Show CLI help. |
Commands:
| Command | Description |
|---|---|
load --config/-c PATH [--source NAME] [--target NAME] [--transform NAME]... |
Validate and execute a source-to-target load job. |
load --dry-run --config/-c PATH [...] |
Validate, read the source, apply transforms, and report target planning without writing. |
load --log-level/-L LEVEL --config/-c PATH [...] |
Run a load with a Loguru level such as DEBUG, INFO, or ERROR. |
run --config/-c PATH [--action NAME] |
Execute a standalone action from actions:. |
run --action-set KEY=VALUE --config/-c PATH [...] |
Override keys on the selected action before execution. |
validate --config/-c PATH [...] |
Validate config shape and print a summary without reading, writing, or executing actions. |
--source and --target are optional when the config defines
exactly one of each; required (and an unknown name is rejected with
a list of available names) when multiple are defined.
--transform is repeatable and optional: each occurrence names a
group from transforms: to concatenate and apply in order. Omit
--transform entirely to run with no transforms.
For run, --action is optional when the config defines exactly
one action and required when multiple actions are defined.
Config files can be YAML (.yaml, .yml) or JSON (.json).
Runtime overrides
The CLI can override selected config values without editing the file. Override values are parsed as YAML, so booleans, numbers, lists, and objects keep their types.
| Flag | Command | Applies to | Example |
|---|---|---|---|
--source-set KEY=VALUE |
load, validate |
selected source | --source-set path=s3://bucket/day=2026-01-01/*.parquet |
--target-set KEY=VALUE |
load, validate |
selected target | --target-set mode=merge --target-set merge_keys='[id]' |
--action-set KEY=VALUE |
run |
selected action | --action-set query='SELECT 1' --action-set params='[123]' |
Dot paths override nested objects:
uv run dataloader-util load -c job.yaml \
--source events --target warehouse \
--source-set format_options.header=false \
--target-set connection.host=db.internal \
--target-set batch_size=10000
uv run dataloader-util run -c job.yaml --action sample_rows \
--action-set connection.host=trino.internal \
--action-set query='SELECT * FROM iceberg.raw.events WHERE ticker = ? LIMIT 10' \
--action-set 'params=[AMD]'
For named action parameters, override individual keys with dotted paths:
uv run dataloader-util run -c job.yaml --action snowflake_sample \
--action-set params.account_id=456 \
--action-set params.status=active
For positional/list action parameters, replace the full list:
uv run dataloader-util run -c job.yaml --action trino_sample \
--action-set 'params=[AMD, 2026-01-01]'
List-index overrides such as params.0=AMD are not currently
supported.
Minimal job config
name: local_csv_to_local_parquet
sources:
events_csv:
kind: local
path: ./data/events.csv
format: csv
fail_on_missing: false # default; skip successfully if no input exists
format_options:
delimiter: ","
header: true
null_values: ["", "null"]
targets:
events_parquet:
kind: local
path: ./output/events.parquet
format: parquet
mode: replace
transforms:
keep_positive:
- kind: sql
sql: "SELECT id, event_ts, user_id, amount FROM <source> WHERE amount > 0"
Configuration reference
Top-level fields:
| Field | Required | Description |
|---|---|---|
name |
No | Optional pipeline name shown in logs and validation output. |
sources |
No | Map of named source configs. At least one is needed for load. |
targets |
No | Map of named target configs. At least one is needed for load. |
transforms |
No | Map of named transform groups; each value is an ordered list of transform steps. |
actions |
No | Map of named standalone actions for run. |
When the config has a single source, a single target, and no
transforms, load runs it with no selection flags. With more than
one source or target, pass --source and/or --target. Pass
--transform zero or more times to select transform groups.
When the config has a single action, run executes it with no
--action flag. With multiple actions, pass --action NAME.
Each named target embeds its own connection: block (required for
postgres/trino/snowflake, omitted for local/s3). Each
query action embeds the connection it executes against. Unknown keys
are rejected at every level.
Sources
Local source
sources:
events:
kind: local
path: ./data/events/*.csv
format: csv
fail_on_missing: false
format_options:
header: true
pathcan be a single file or a DuckDB-supported glob.- Exact paths and globs are checked before reading.
- If no file matches and
fail_on_missingis omitted orfalse, the run skips successfully with zero rows written. - Set
fail_on_missing: trueto fail before transforms or writes when no local input exists.
S3 source
sources:
events:
kind: s3
path: s3://my-bucket/raw/events/*.parquet
format: parquet
fail_on_missing: false
format_options:
hive_partitioning: true
region: us-east-1
endpoint_url: http://localhost:9000 # optional, for MinIO/LocalStack
access_key_id: ${AWS_ACCESS_KEY_ID} # optional
secret_access_key: ${AWS_SECRET_ACCESS_KEY} # optional
- S3 reads use DuckDB's
httpfsextension. - Exact object keys and glob patterns are checked before reading.
- Missing objects or globs with no matching objects skip successfully by default; set
fail_on_missing: trueto fail before transforms or writes. - Permission denied, bucket-not-found, credential, and network failures are always treated as errors, not as empty input.
- If explicit keys are omitted, DuckDB/AWS credential discovery is used.
- Set
endpoint_urlfor S3-compatible systems such as MinIO or LocalStack.
Formats
| Format | Reader | Options |
|---|---|---|
csv |
DuckDB read_csv_auto |
delimiter, header, null_values, compression (none, gzip, zstd) |
jsonl |
DuckDB read_json_auto with newline-delimited format |
compression (none, gzip, zstd) |
parquet |
DuckDB read_parquet |
columns, hive_partitioning |
Examples:
format: csv
format_options:
delimiter: "|"
header: true
null_values: ["", "NULL"]
compression: gzip
format: parquet
format_options:
columns: [id, event_ts, amount]
hive_partitioning: true
Transforms
A transforms: entry maps a group name to an ordered list of
transform steps. The CLI concatenates the steps from each
--transform NAME flag in the order the flags are given. Each
transform reads the previous DuckDB view and registers a new
temporary view.
SQL transform
Use <source> as the placeholder for the current input view.
transforms:
keep_positive:
- kind: sql
sql: "SELECT id, amount * 100 AS amount_cents FROM <source> WHERE amount > 0"
Filter transform
The where value is the expression only; do not include WHERE.
transforms:
active_only:
- kind: filter
where: "amount > 0 AND status = 'active'"
Rename transform
Mapping is old_name: new_name. Unmapped columns pass through unchanged.
transforms:
camel_case:
- kind: rename
mapping:
event_ts: eventTs
user_id: userId
Cast transform
Types are DuckDB SQL type names.
transforms:
cast_clean:
- kind: cast
columns:
user_id: VARCHAR
amount: DOUBLE
event_ts: TIMESTAMP
Load job selection patterns
Single source and target, no transforms:
uv run dataloader-util load -c examples/job_local_to_local.yaml
Multiple named buckets:
uv run dataloader-util load -c examples/job_multi_source_target.yaml \
--source events_csv \
--target events_postgres \
--transform keep_positive \
--transform camel_case
Validation with runtime target overrides:
uv run dataloader-util validate -c job.yaml \
--source data \
--target iceberg \
--target-set table=iceberg.raw.events \
--target-set mode=merge \
--target-set 'merge_keys=[id, account_id]'
Dry run without writing:
uv run dataloader-util load -c job.yaml --dry-run \
--source data --target warehouse --transform clean
Targets
Targets live in the top-level targets: map, keyed by name. Each
target embeds its own connection: block (required for
postgres/trino/snowflake, omitted for local/s3).
Local file target
targets:
events_parquet:
kind: local
path: ./output/events.parquet
format: parquet
mode: replace
- Supported output formats:
parquet,csv. - Parent directories are created automatically.
- Use
mode: replace; local output is written with DuckDBCOPYto the configured file.
S3 target
targets:
archive_parquet:
kind: s3
path: s3://my-bucket/curated/events.parquet
format: parquet
mode: replace
region: us-east-1
endpoint_url: http://localhost:9000 # optional, for MinIO/LocalStack
access_key_id: ${AWS_ACCESS_KEY_ID} # optional
secret_access_key: ${AWS_SECRET_ACCESS_KEY} # optional
- Supported output formats:
parquet,csv. - The target format is independent from the source format, so a CSV or JSONL source can be archived as Parquet by setting
format: parqueton the target. - S3 writes use DuckDB's
httpfsextension and the same optional credential/endpoint fields as S3 sources. - If explicit keys are omitted, DuckDB/AWS credential discovery is used.
Postgres target
targets:
warehouse:
kind: postgres
table: analytics.events
mode: replace
batch_size: 50000
connection:
kind: postgres
host: localhost
port: 5432
user: etl
password: ${POSTGRES_PASSWORD}
database: analytics
schema: public
- Table names can be
tableorschema.table; a bare table defaults topublic. replacedrops and recreates the table.appendcreates the table if missing and inserts rows.mergecreates the table with a primary key if missing, stages rows in a temp table, then usesINSERT ... ON CONFLICT.- Bulk loading uses
psycopgCOPY FROM STDIN.
Trino target
targets:
events_iceberg:
kind: trino
table: iceberg.raw.events
mode: append
batch_size: 5000
connection:
kind: trino
host: localhost
port: 8080
user: etl
catalog: iceberg
schema: raw
- Table names must be
catalog.schema.table. - The target schema is created if missing.
- Writes use a temporary table plus server-side insert or merge.
- Iceberg support is provided through Trino's Iceberg connector configuration on the Trino server.
- To register existing S3 Parquet files as an Iceberg table without scanning rows, use the
trino_iceberg_registeraction instead.
Snowflake target
targets:
events_merged:
kind: snowflake
table: RAW.EVENTS
mode: merge
merge_keys: [id]
batch_size: 100000
connection:
kind: snowflake
account: ${SNOWFLAKE_ACCOUNT}
user: ${SNOWFLAKE_USER}
password: ${SNOWFLAKE_PASSWORD}
warehouse: COMPUTE_WH
database: ANALYTICS
schema: RAW
role: ${SNOWFLAKE_ROLE}
- Table names can be
tableorschema.table; the database comes from the connection. - Writes use
snowflake-connector-pythonwrite_pandas. mergestages data into a temporary table and runs a SnowflakeMERGEstatement.- Snowflake identifiers are validated as unquoted identifier names.
Actions
Actions live in the top-level actions: map, keyed by name. They are
standalone units of work executed with dataloader-util run; they do
not read from sources:, apply transforms:, or write to targets:.
Use actions for parameterized warehouse queries, table maintenance,
metadata operations, checks, or small operational SQL tasks.
Action CLI usage
Run the only action in a config:
uv run dataloader-util run -c examples/job_trino_query_action.yaml
Run a named action:
uv run dataloader-util run -c job.yaml --action sample_events
Override action config values at runtime:
uv run dataloader-util run -c job.yaml --action sample_events \
--action-set connection.host=trino.internal \
--action-set query='SELECT * FROM iceberg.raw.events WHERE ticker = ? LIMIT 5' \
--action-set 'params=[AMD]'
run prints a concise summary with the action kind, affected row
count when reported by the driver, returned row count, and up to the
first 20 returned rows.
Trino query action
actions:
sample_events:
kind: trino_query
connection:
kind: trino
host: trino.example.com
port: 8080
user: etl_user
catalog: iceberg
schema: raw
# password: ${TRINO_PASSWORD} # optional; enables basic auth
query: |
SELECT *
FROM iceberg.raw.events
WHERE ticker = ?
LIMIT 10
params:
- AMD
Notes:
- Trino action connections reuse the same
TrinoConnectionfields as Trino targets. - Positional parameters with
?placeholders and list-shapedparamsare the safe/default style for the Trino Python client. - Dict-shaped
paramsare accepted by config validation, but named placeholder support depends on the Trino driver/server behavior; prefer positional params unless verified in your environment. - Use
--action-set 'params=[AMD]'to replace positional params at runtime.
Trino Iceberg register action
Register existing S3 Parquet files as an Iceberg table through Trino without
scanning any rows. Builds the table DDL from an explicitly declared
schema (column name → Trino type) and runs
ALTER TABLE ... EXECUTE add_files(...) to attach the files.
actions:
register_events:
kind: trino_iceberg_register
connection:
kind: trino
host: trino.example.com
port: 8080
user: etl_user
catalog: iceberg
schema: raw
table: iceberg.raw.events
mode: append # or replace
location: s3://bucket/events/
schema:
id: BIGINT
event_ts: TIMESTAMP
payload: VARCHAR
format: PARQUET
recursive_directory: true # true | false | fail
create_table_if_missing: true
table_location: s3://bucket/iceberg/raw/events/ # optional
partitioning: [] # optional, e.g. [day(event_ts)]
Notes:
- Requires the Trino Iceberg catalog property
iceberg.add-files-procedure.enabled=true. locationmust be ans3://URI reachable by Trino workers.schemais an explicit column-name → Trino-type map and is required. Column order follows the map order. Types are passed through verbatim (including parametric types likeVARCHAR(255)orDECIMAL(18,2)); Trino validates them at table-creation time.- Trino does not validate Parquet file schemas during
add_files; the declaredschemamust match the files, or later queries will fail. mode: replacedrops and recreates the table before adding files;mode: appendcreates the table if missing (or requires it to exist whencreate_table_if_missing: false).recursive_directorycontrols how files underlocationare discovered:true,false, orfail(abort if subdirectories exist).- Use
--action-set schema.id=BIGINTto override a column type at runtime.
Snowflake query action
actions:
sample_events:
kind: snowflake_query
connection:
kind: snowflake
account: ${SNOWFLAKE_ACCOUNT}
user: ${SNOWFLAKE_USER}
password: ${SNOWFLAKE_PASSWORD}
warehouse: COMPUTE_WH
database: ANALYTICS
schema: RAW
role: ${SNOWFLAKE_ROLE}
query: |
SELECT *
FROM EVENTS
WHERE ACCOUNT_ID = %(account_id)s
LIMIT 10
params:
account_id: 123
Notes:
- Snowflake action connections reuse the same
SnowflakeConnectionfields as Snowflake targets. - Named parameters are supported with dict-shaped
paramsand Snowflake connector placeholder syntax such as%(account_id)s. - Override one named param at runtime with
--action-set params.account_id=456. - Positional/list params can also be supplied when using positional placeholder syntax supported by the connector.
Action-only config
A config can contain only actions: if it is intended for run:
name: warehouse_actions
actions:
trino_sample:
kind: trino_query
connection:
kind: trino
host: trino.example.com
user: etl
catalog: iceberg
schema: raw
query: "SELECT count(*) AS rows FROM iceberg.raw.events"
Validate it without executing the query:
uv run dataloader-util validate -c actions.yaml
Then run it:
uv run dataloader-util run -c actions.yaml --action trino_sample
Write modes
| Mode | Local | Postgres | Trino | Snowflake |
|---|---|---|---|---|
replace |
Write/overwrite output file | Drop and recreate table | Drop and recreate table | Drop table, then write with auto-create |
append |
Not intended for local files | Create table if needed, then append | Create table if needed, then append | Create table if needed, then append |
merge |
Not supported for local files | Upsert by merge_keys with ON CONFLICT |
MERGE INTO ... USING |
MERGE INTO ... USING |
For merge, target.merge_keys is required and each key must exist in the transformed source columns.
Missing source behavior
All source kinds support fail_on_missing:
sources:
daily_events:
kind: local # also works for kind: s3
path: ./data/daily/*.csv
format: csv
fail_on_missing: false
Default behavior is false: if the source exact path/object is absent or the glob matches no files/objects, the run logs source has no data; run skipped, applies no transforms, writes nothing, exits successfully, and reports zero rows written with a skipped result indicator.
Set fail_on_missing: true for strict jobs that should fail when no input exists.
Environment variable substitution
Any string value can contain ${VAR}. Placeholders are resolved before schema validation.
targets:
warehouse:
kind: postgres
table: analytics.events
connection:
kind: postgres
password: ${POSTGRES_PASSWORD}
Rules:
- Missing environment variables raise
ConfigError. - Use environment variables for credentials and account-specific values.
- Do not commit real secrets to config files.
Included examples
| File | Purpose |
|---|---|
examples/job_local_to_local.yaml |
Local CSV to local Parquet with SQL filtering. |
examples/job_local_to_local_with_transforms.yaml |
Local CSV to Parquet using filter, rename, and cast transforms. |
examples/job_local_to_postgres.yaml |
Local CSV to Postgres replace load. |
examples/job_local_to_postgres_merge.yaml |
Local CSV to Postgres upsert by key. |
examples/job_s3_to_local.yaml |
S3 Parquet to local Parquet. |
examples/job_s3_to_postgres_merge.yaml |
S3 Parquet to Postgres merge. |
examples/job_s3_to_snowflake.yaml |
S3 Parquet to Snowflake append. |
examples/job_local_to_trino_iceberg.yaml |
Local CSV to an Iceberg table through Trino. |
examples/job_s3_to_trino_iceberg_metadata.yaml |
Register existing S3 Parquet files as Iceberg metadata through Trino. |
examples/job_trino_query_action.yaml |
Standalone parameterized Trino query action for dataloader-util run. |
examples/job_snowflake_query_action.yaml |
Standalone named-parameter Snowflake query action for dataloader-util run. |
examples/job_multi_source_target.yaml |
One file with multiple sources, targets, and transform groups; pick with --source / --target / --transform. |
Local integration services
The repository includes docker-compose.yml for local integration dependencies:
docker compose up -d
just test-integration
docker compose down
Services include:
- MinIO on
localhost:9000with console onlocalhost:9001 - Postgres on
localhost:5432(etl/etlpass/analytics) - Trino on
localhost:8080with the memory catalog
Snowflake integration tests are gated by SNOWFLAKE_* environment variables and require access to a real account.
Development
Common tasks are defined in justfile:
just # list available tasks
just dev # uv sync
just test # unit tests only
just test-integration
just test-all
just test-cov
just lint
just fmt # format check
just fmt-write # apply formatting
just types
just check # lint + types + unit tests
just fix # ruff auto-fix + format
Build artifacts can be produced with:
uv build
Troubleshooting
config file not found: check the--configpath and current working directory.unsupported config file extension: use.yaml,.yml, or.json.- Validation errors: remove unknown keys, ensure target/action and connection kinds match, and provide
merge_keysfor merge jobs. multiple actions defined; --action is required: pass--action NAME, or keep only one entry underactions:.--action-setdid not update a positional param: replace the whole list with--action-set 'params=[value1, value2]'; list-index overrides are not supported.- Trino named params fail: use positional
?placeholders and list-shapedparams; named parameter behavior is driver-dependent. - Run skipped with
source has no data: the selected source path/object is absent or its glob matched nothing. This is successful by default; setfail_on_missing: trueto make it an error. - S3 read failures: verify region, credentials, bucket path, and
endpoint_urlfor S3-compatible services. Permission, bucket, credential, and network failures always error. - Trino table errors: use a fully qualified
catalog.schema.tabletarget name and verify the catalog exists on the server. - Snowflake identifier errors: use simple unquoted-style database, schema, and table names.
License
TBD