Files and Delta (s3, delta, binary)

This page covers the three connectors that read and write files: by path (s3), as Delta Lake tables (delta), and as raw bytes (binary). Use it when a task reads or writes files in cloud storage, HDFS or a local folder.

• s3: any path (S3, ADLS, GCS, HDFS, local). Supports Parquet, CSV, JSON, ORC, Avro and Delta. It also runs on the pandas backend (no Spark, no Java).
• delta: Delta Lake tables, accessed by path or by table name. Needs Spark.
• binary: raw files (PDFs, images), one row per file. Read only. See Fields: binary.

---

s3: path-based reads and writes

Read

CONFIG:
  inputs:
    raw_events:
      format: s3
      path: s3://my-bucket/raw/events/dt=2024-06-01/
      options:
        mergeSchema: "true"

Default file format is Parquet. Choose another with file_format (format is already taken: it picks the connector). Reader options go under options:

    csv_data:
      format: s3
      path: s3://my-bucket/exports/customers.csv
      file_format: csv
      options:
        header: "true"
        inferSchema: "true"

To skip schema inference, give schema as a DDL string: schema: "id BIGINT, name STRING".

CSV with text that holds quotes or line breaks

Ubunye reads CSV the way Spark does, on both backends, so the same file gives the same rows. Spark's default escape character is a backslash. Most CSV files written by pandas, Excel or a database escape a quote by doubling it (""), and hold line breaks inside quoted text. Read such a file with Spark's defaults and some rows split in the wrong place: values land in the wrong columns, and inferSchema then reads a number column as text, without a warning. Tell the reader how the file is written:

      options:
        header: "true"
        inferSchema: "true"
        multiLine: "true"   # quoted text may span lines
        escape: '"'         # a quote inside quoted text is written as ""

A sign you need this: an expectation such as between says a number column is text, or the row count differs from pandas.read_csv on the same file. A UTF-8 byte order mark at the start of a file is dropped, as Spark does.

Write

CONFIG:
  outputs:
    clean_events:
      format: s3
      path: s3://my-bucket/clean/events/
      mode: overwrite
      partitionBy: [event_date]     # a top-level list, not an option
      options:
        compression: snappy

---

delta: Delta Lake

Read by path

CONFIG:
  inputs:
    transactions:
      format: delta
      path: s3://my-bucket/delta/transactions/

Read by table name

    transactions:
      format: delta
      table: main.finance.transactions    # or db_name + tbl_name

Time travel

    transactions:
      format: delta
      path: s3://my-bucket/delta/transactions/
      version_as_of: 12                   # or timestamp_as_of: "2026-01-01"

options: {versionAsOf: 12} (or timestampAsOf, in any case) works the same.

One version per run

A Delta read that names no version is pinned to the version the table is at when the task reads it (versionAsOf, taken from DESCRIBE HISTORY ... LIMIT 1). Every output and the run record then see that one snapshot, even if another job appends while the task runs. Without this pin, a Delta frame would read the latest version each time Spark computes it, so two outputs of one input could come from two versions. This applies to format: delta and to format: s3 with file_format: delta. A sql: input and a catalog table read through hive or unity are not pinned. If the version cannot be read (no Delta SQL extension, no permission to see the history), the read is left unpinned, as before. The run record says which it was (pinned_by: engine or config).

Write

CONFIG:
  outputs:
    predictions:
      format: delta
      path: s3://my-bucket/delta/predictions/    # or table: main.fraud.predictions
      mode: overwrite
      options:
        mergeSchema: "true"
        overwriteSchema: "true"

format: delta and format: s3 + file_format: delta are equivalent; use whichever reads better. The delta connector addresses a table (by name or path); the s3 connector addresses a path that happens to hold Delta files. Both support all seven write modes.

Write modes

All seven write modes work on delta, and on s3 with file_format: delta. The Inputs & Outputs page owns how each mode behaves. One example, a MERGE (upsert) on Delta:

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

For backfills that replace only some partitions, see overwrite_partitions. Its partitionBy form also runs on the pandas backend, with the same folders and the same result as Spark; replace_where needs Delta, so Spark. See the pandas backend.

Appends that land once, and reruns that correct

A Delta append of a named batch (a run with -dt or a --var) lands once. A run killed or failed after its commit leaves the rows; the next run of the batch commits nothing (Delta's idempotent write, keyed by the task, the output and the batch). If the table did not change, the run says the batch was already there.

To make a rerun correct a bad batch, name the batch's rows with replace_where:

    payments:
      format: delta
      table: main.billing.payments
      mode: append
      replace_where: "month = '{{ dt }}'"

Each run then replaces the rows the predicate selects, in one commit: a retry after a crash, and --rerun of a finished batch, write the batch from the current data. Delta refuses data outside the predicate, so a wrong predicate fails the run and deletes nothing. A second plain run of a finished batch is still refused (pass --rerun). Without replace_where, --rerun of a finished batch appends it again and says so. See ADR 008.

---

Fields: s3

Field	Type	Required	Description
format	"s3"	Yes	Selects this connector
path	string	Yes	S3 / ADLS / GCS / HDFS / local path
file_format	string	No	parquet (default), delta, csv, json, orc, avro
schema	string	No	Reads only: a DDL schema string; skips inference
mode	see Write modes	No	Default append. All seven modes supported; merge, insert_new and replace_where require file_format: delta
partitionBy	list	No	Partition columns; required for mode: overwrite_partitions unless replace_where is set
merge_keys	list | string	Conditional	Required for mode: merge
replace_where	string	No	Delta predicate for mode: overwrite_partitions; on a Delta append, the batch's rows, replaced by each run
options	dict	No	Spark reader/writer options (e.g. mergeSchema)

Fields: delta

Field	Type	Required	Description
format	"delta"	Yes	Selects this connector
path	string	Conditional	Delta table path (required unless a table is named)
table	string	Conditional	Table name, e.g. main.fraud.claims
db_name + tbl_name	string	Conditional	The table name in two parts, instead of table
sql	string	No	Spark SQL query, reads only. Give path or a table name too
mode	see Write modes	No	Default append. All seven modes supported
partitionBy	list	No	Partition columns
merge_keys	list | string	Conditional	Required for mode: merge
replace_where	string	No	Predicate for mode: overwrite_partitions; on an append, the batch's rows, replaced by each run (above)
version_as_of	int	No	Time travel by version (reads only)
timestamp_as_of	string	No	Time travel by timestamp (reads only)
options	dict	No	Spark reader/writer options

Fields: binary

Reads raw files (PDFs, images, audio) one row per file, with columns path, modificationTime, length, content. Spark's binaryFile source is read-only: format: binary in CONFIG.outputs is rejected at config load.

Field	Type	Required	Description
format	"binary"	Yes	Selects this connector
path	string	Yes	Directory or file path
path_glob_filter	string	No	Only read matching files, e.g. "*.pdf"
recursive	bool	No	Descend into sub-directories
options	dict	No	Spark reader options

---

Spark configuration for S3

ENGINE:
  spark_conf:
    spark.hadoop.fs.s3a.access.key: "{{ env.AWS_ACCESS_KEY_ID }}"
    spark.hadoop.fs.s3a.secret.key: "{{ env.AWS_SECRET_ACCESS_KEY }}"
    spark.hadoop.fs.s3a.endpoint: "s3.amazonaws.com"
    # For Delta Lake on open-source Spark. Match the Delta build to your Spark:
    # (Spark 4.0: delta-spark_2.13:4.0.0; Spark 3.5: delta-spark_2.12:3.2.0)
    spark.jars.packages: "io.delta:delta-spark_2.13:4.0.0"
    spark.sql.extensions: "io.delta.sql.DeltaSparkSessionExtension"
    spark.sql.catalog.spark_catalog: "org.apache.spark.sql.delta.catalog.DeltaCatalog"

spark.jars.packages and spark.sql.extensions are fixed when the Spark session starts: where the platform starts it (EMR, Dataproc, spark-submit), set them there. See Spark keys fixed at session start. The Python package is pip install "ubunye-engine[delta]".

The two key lines template environment variables because secret:// references are not read in ENGINE.spark_conf (see Secrets). Better still, leave them out and let Spark use the platform's login (an instance role, workload identity).

On Databricks, Delta and S3 are pre-configured, so no extra Spark settings are needed.

---

Partitioned paths with Jinja

  outputs:
    daily_events:
      format: delta
      path: "s3://my-bucket/delta/events/dt={{ dt | default('2024-01-01') }}/"
      mode: overwrite