Skip to content

S3 / Delta Connector

Two closely related connectors handle path-based and Delta Lake storage.

  • s3 — any Spark-readable path (S3, HDFS, ADLS, local). Supports Parquet, CSV, JSON, ORC, Avro.
  • delta — Delta Lake tables, accessed by path or by three-part table name.

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 format is Parquet. Override with options.format or by specifying a Spark data source format in options:

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

Write

CONFIG:
  outputs:
    clean_events:
      format: s3
      path: s3://my-bucket/clean/events/
      mode: overwrite
      options:
        partitionBy: event_date
        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"

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 six write modes.

MERGE (upsert)

Matched rows are updated on merge_keys, unmatched rows inserted. The first run creates the table (there is nothing to merge into yet).

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

Partition overwrite (backfills)

Replace only the partitions in the incoming DataFrame — the rest of the table is untouched. Use partitionBy for Spark's dynamic partition overwrite, or replace_where for a Delta predicate.

    daily_sales:
      format: s3
      path: s3://my-bucket/sales/
      mode: overwrite_partitions
      partitionBy: [dt]

    # Delta predicate form — replaces exactly what the expression matches
    daily_sales_delta:
      format: s3
      path: s3://my-bucket/delta/sales/
      file_format: delta
      mode: overwrite_partitions
      replace_where: "dt = '{{ dt }}'"

Without partitionBy or replace_where this mode is refused — dynamic overwrite of an unpartitioned target is a full-table wipe.

Fields — s3

Field Type Required Description
format "s3" Yes Selects this connector
path string Yes S3 / HDFS / local path
file_format string No parquet (default), delta, csv, json, orc, avro
mode see Write modes No Default append. All six modes supported; merge 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
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 table set)
table string Conditional Table name, or db_name + tbl_name
sql string Conditional Spark SQL query (reads only)
mode see Write modes No Default append. All six 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
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:
    spark.jars.packages: "io.delta:delta-core_2.12:2.4.0"
    spark.sql.extensions: "io.delta.sql.DeltaSparkSessionExtension"
    spark.sql.catalog.spark_catalog: "org.apache.spark.sql.delta.catalog.DeltaCatalog"

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

Partitioned paths with Jinja

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