Inputs & Outputs¶

This page is for anyone writing a task's config.yaml. It shows how to say what a task reads (CONFIG.inputs) and writes (CONFIG.outputs), and it is the home page for write modes: how an output treats data already in its target.

Each entry has a name you choose and a block that names a connector with format:. A connector is the built-in code that reads or writes one kind of storage (see Connectors).

Structure¶

CONFIG:
  inputs:
    <logical_name>:
      format: <format_type>
      # format-specific fields...

  outputs:
    <logical_name>:
      format: <format_type>
      mode: append | overwrite | errorifexists | ignore | merge | insert_new | overwrite_partitions
      # format-specific fields...

At least one input and one output are required.
Logical names are arbitrary; they are passed to your transform as dict keys.

Supported formats¶

`format`	Direction	Connector
`hive`	read / write	Hive Metastore tables via Spark
`jdbc`	read / write	Any JDBC-compatible database
`unity`	read / write	Databricks Unity Catalog
`s3`	read / write	Any path: S3, ADLS, GCS, HDFS, local (Parquet, CSV, JSON...)
`delta`	read / write	Delta Lake tables by path or name
`binary`	read	Raw binary files
`rest_api`	read / write	HTTP REST endpoints

format may also name any connector installed as a plugin. ubunye plugins lists what is installed; an unknown format is refused with that list.

Common fields¶

These fields apply to most formats:

Field	Type	Description
`format`	string	Required. One of the format types above.
`options`	dict	Spark reader/writer options (e.g. `header`, `delimiter`).
`mode`	see Write modes	Write mode (outputs only).
`merge_keys`	list or string	Key columns for `mode: merge` and `mode: insert_new` (outputs only).
`replace_where`	string	Delta predicate for `mode: overwrite_partitions` (outputs only). On a Delta or JDBC `append`, the rows of the run's batch: each run replaces them, so a rerun corrects the batch instead of adding it twice.

Format-specific fields¶

`hive`¶

format: hive
db_name: raw             # required if sql is not set
tbl_name: claims         # required if sql is not set
sql: "SELECT ..."        # alternative to db_name + tbl_name (inputs only)

As an output, hive needs db_name + tbl_name.

`jdbc`¶

format: jdbc
url: "jdbc:postgresql://host:5432/db"   # required
table: public.claims                    # required unless sql is set (always for outputs)
sql: "SELECT * FROM public.claims"      # alternative to table (inputs only)
user: "{{ env.DB_USER }}"
password: "secret://env/DB_PASS"        # see Secrets
options:
  fetchsize: "10000"
  partitionColumn: id
  lowerBound: "1"
  upperBound: "1000000"
  numPartitions: "8"

`unity`¶

format: unity
table: main.fraud.claims       # three-part Unity Catalog name
# or:
sql: "SELECT ..."              # inputs only
# outputs also take catalog + schema + tbl_name instead of table

A name that comes from a variable goes through | ident, which quotes it once (a catalog like aws-x-catalog needs backticks in SQL): sql: "SELECT * FROM {{ env.TELM_READ_CATALOG | ident }}.raw.trips". See SQL names.

`s3` / `delta`¶

# S3: path-based; file_format picks the file type (default parquet)
format: s3
path: s3://my-bucket/data/claims/
file_format: csv
options:
  header: "true"

# Delta: by path
format: delta
path: s3://my-bucket/delta/claims
mode: overwrite

# Delta: by table name
format: delta
table: main.fraud.claims
mode: append

`binary`¶

format: binary
path: /mnt/raw/documents/
path_glob_filter: "*.pdf"   # optional
recursive: true             # optional

`rest_api`¶

format: rest_api
url: "https://api.example.com/v1/records"
auth:
  type: bearer
  token: "secret://env/API_TOKEN"
pagination:
  type: cursor
  cursor_response_key: next_cursor
headers:
  Accept: application/json

See REST API Connector for the full REST API reference.

Connector keys are checked¶

An input or output block carries whatever keys its connector reads, and the whole block is passed to the connector. This is how the REST API connector receives auth, pagination and headers without schema changes.

Each built-in connector declares the keys it reads, so any other key is refused by ubunye validate, with the closest real one:

outputs.clean: 'paht' is not a setting of 's3'; did you mean 'path'?

Each connector also says what it requires (s3 needs path, jdbc needs url, and so on), checked the same way. Options for Spark itself go under options. A third-party connector that declares no keys accepts any key.

Write modes¶

The write mode (mode: on an output) says what happens to data already in the target. The first four are Spark's own save modes. merge, insert_new and overwrite_partitions are added by Ubunye: Spark has no save mode for them.

Mode	Behaviour
`append`	Insert new rows without touching existing data
`overwrite`	Drop existing data and replace entirely
`errorifexists`	Fail if the target already exists (alias: `error`)
`ignore`	Do nothing if the target already exists
`merge`	Delta MERGE (upsert) on `merge_keys`. Creates the target on first run
`insert_new`	Insert only rows whose `merge_keys` are not in the target yet; existing rows never change (first write wins). Delta. Creates the target on first run
`overwrite_partitions`	Replace only the partitions present in the DataFrame; leave the rest of the table intact

Not every connector can honour every mode. Asking for one it cannot do raises SinkWriteError before any rows are written. It is never silently downgraded.

ubunye validate and ubunye plan both catch this before a run, with the same message:

  problem: outputs.predictions: Write mode 'merge' is not supported by the 'jdbc' connector.

Connector	Supported modes	Default
`delta`	all seven	`append`
`unity`	all seven	`append`
`s3`	all seven (`merge`, `insert_new` and `replace_where` need `file_format: delta`)	`append`
`hive`	all seven (`merge` and `insert_new` need `file_format: delta`)	`append`
`jdbc`	native four only: no `merge` / `insert_new` / `overwrite_partitions`	`append`
`rest_api`	`append` (a REST sink has no table to merge into)	`append`
`binary`	none: read-only, rejected in `CONFIG.outputs`	none

Every writer defaults to append: the mode that cannot destroy data you did not mean to destroy. Set mode explicitly when you want anything else.

`merge`¶

Requires merge_keys: the columns that identify a row. Matched rows are updated, unmatched rows inserted. On the first run the target does not exist yet, so the engine creates it with a plain overwrite instead of a MERGE.

    customer_features:
      format: s3
      path: s3://my-bucket/delta/features/
      file_format: delta
      mode: merge
      merge_keys: [id, dt]        # or options.merge_keys: "id,dt"

`insert_new`¶

Insert only the rows whose merge_keys are not in the target yet. A row whose keys are already there is left exactly as it is, even if the new row differs: the first write of a key wins, and running the same batch again adds nothing. Use it for facts that never change once recorded (events, first-seen dates, an audit trail), where append would duplicate on a rerun and merge would overwrite history.

    trips:
      format: unity
      table: "{{ env.TELM_WRITE_CATALOG | ident }}.gold.trips"
      mode: insert_new
      merge_keys: [trip_id]       # or options.merge_keys: "trip_id"

merge_keys are required, and the target must be Delta (format: delta or unity, or file_format: delta on s3 / hive). ubunye validate and ubunye plan refuse anything else before a run.
On Spark it is one Delta MERGE with only WHEN NOT MATCHED THEN INSERT *, on a table or a path. No existing file is rewritten.
Keys are compared null-safe (<=>): a row whose key is null is inserted once, not on every run.
Rows that share a key inside one batch are all inserted, as Delta does. Drop duplicates in your transform if a batch may carry them.
On the first run the target is created. A path that holds files but is not a Delta table is refused, never replaced or written over.
The pandas backend, jdbc and rest_api refuse it with a clear message. pandas cannot write Delta; a read-then-append on plain files could not promise "changes nothing" if two runs overlap.

`overwrite_partitions`¶

The backfill mode: re-run one day without wiping the rest of the table. Three ways to say which partitions to replace:

    # 1. Dynamic: replace whatever partitions the DataFrame contains
    daily_sales:
      format: s3
      path: s3://my-bucket/sales/
      mode: overwrite_partitions
      partitionBy: [dt]

    # 2. Predicate: replace exactly what the expression matches (Delta only)
    daily_sales:
      format: delta
      path: s3://my-bucket/delta/sales/
      mode: overwrite_partitions
      replace_where: "dt = '{{ dt }}'"

    # 3. A table: INSERT OVERWRITE of just the incoming partitions
    daily_sales:
      format: hive
      db_name: clean
      tbl_name: sales
      mode: overwrite_partitions
      partitionBy: [dt]

Without partitionBy or replace_where, dynamic overwrite silently degrades to a full-table wipe, so the engine refuses it instead. ubunye plan reports this before a run (validate does not):

  problem: outputs.daily_sales: Write mode 'overwrite_partitions' requires 'partitionBy' or 'replace_where'.

The partitionBy form also runs on the pandas backend; replace_where needs Delta, so Spark.

`replace_where` on an append¶

On a Delta or JDBC append, replace_where names the rows of the run's batch, so a rerun replaces the batch instead of adding it twice. Each connector page owns the details: Delta appends and JDBC appends.

Under the hood

Checking a mode against what a connector supports is done once, the same way on every backend. Running merge, insert_new and overwrite_partitions is Spark work (MERGE INTO, dynamic partition overwrite), done by the Spark backend.

For overwrite_partitions the engine picks the mechanism the target actually supports: Delta's own write option for Delta, INSERT OVERWRITE for an existing table, and Spark's partitionOverwriteMode=DYNAMIC for a path. Each route is covered by an integration test against a real Spark session.

Inputs & Outputs¶

Structure¶

Supported formats¶

Common fields¶

Format-specific fields¶

hive¶

jdbc¶

unity¶

s3 / delta¶

binary¶

rest_api¶