Delta Lake

Sink The Delta Lake connector writes streaming data to Delta Lake tables backed by object storage. It produces Parquet data files and commits them to the Delta transaction log, giving you ACID guarantees and time-travel queries over streaming output. The connector supports S3, GCS, Azure Blob Storage, and local filesystem paths. Delta Lake is sink-only - there is no source mode. The connector has no connection profile (no separate “connection” config) because all storage credentials are provided inline on the table config via storage_options.

Table Path and Storage

The only required field beyond type is the table path. Ingest Engine parses the URI scheme to determine the storage backend - s3://, gs://, az:// (or abfs://), or a local filesystem path.

table_type:
  type: sink
  sink_config:
    path: s3://my-bucket/delta-tables/events/

path (required) - URI of the Delta Lake table location. This is where Parquet data files and the _delta_log/ directory live. Ingest Engine calls BackendConfig::parse_url at validation time - if the URI scheme is unrecognized or the path is malformed, the pipeline fails immediately at startup. Supported schemes: s3://, gs://, az://, and local paths (e.g., /data/delta/events).

Storage Options

The storage_options map passes credentials and configuration directly to the underlying object store client. Keys are parsed as AmazonS3ConfigKey, GoogleCloudStorageConfigKey, or AzureConfigKey depending on the path scheme. If no access key is provided for S3, Ingest Engine falls back to its built-in credential provider (IAM roles, environment variables, instance metadata). AWS S3:

storage_options:
  aws_access_key_id: AKIAIOSFODNN7EXAMPLE
  aws_secret_access_key: wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY
  aws_region: us-east-1
  aws_endpoint: https://s3.us-east-1.amazonaws.com  # optional, for custom endpoints

If aws_access_key_id is omitted, Ingest Engine uses its own credential provider which checks AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY environment variables, then the AWS credential chain (instance profile, ECS task role, etc.). When a custom aws_endpoint is set, Ingest Engine automatically enables path-style access and allows HTTP (useful for MinIO or LocalStack). Google Cloud Storage:

storage_options:
  google_service_account_key: <GCP_SERVICE_ACCOUNT_JSON>
  # OR
  google_service_account_path: /path/to/credentials.json

Azure Blob Storage:

storage_options:
  azure_storage_account_name: mystorageaccount
  azure_storage_account_key: YOUR_ACCOUNT_KEY
  # OR use SAS token
  azure_storage_sas_token: YOUR_SAS_TOKEN

The key names in storage_options are the same keys accepted by the object_store Rust crate’s builder for each cloud provider. Refer to the object_store documentation for the complete list.

Rolling Policy

The rolling policy controls when Ingest Engine closes the current Parquet file and starts writing a new one. Without any rolling policy, files grow until a checkpoint triggers a close. You can set size limits, time intervals, inactivity timeouts, or any combination - whichever condition triggers first wins.

rolling_policy:
  file_size_bytes: 134217728
  interval_seconds: 300

file_size_bytes (optional, integer) - Roll after the file reaches this many bytes. Ingest Engine tracks bytes written to the multipart upload and rolls when the threshold is crossed. A value of 134217728 (128 MB) is a good starting point for most workloads. Larger files mean fewer Delta log entries and faster reads; smaller files mean lower commit latency.
interval_seconds (optional, integer, min: 1) - Roll after this many seconds have elapsed since the first write to the current file. This is a wall-clock duration measured from when the file was opened, not from the last write. Set to 300 (5 minutes) for a balance between freshness and file count.
inactivity_seconds (optional, integer, min: 1) - Roll after this many seconds of inactivity (no new data written). Useful for bursty workloads where you want files to close promptly when traffic drops.

In addition to these user-configured policies, Ingest Engine always enforces a hard limit on the number of multipart upload parts per file (default 1000, configurable via multipart.max_parts). Once the part limit is reached the file is rolled regardless of other settings. When a time_pattern is set on partitioning (see below), Ingest Engine also adds a watermark-based rolling policy: if the pipeline watermark advances past the current file’s time partition boundary, the file is closed. This prevents late data from keeping old partition files open indefinitely.

Partitioning

Partitioning controls how data is organized into directory paths within the Delta table. Ingest Engine uses Iceberg-style partition transforms internally - each record is evaluated against the partition fields and routed to a directory named field=value.

partitioning:
  fields:
    - name: event_date
      transform: identity
  shuffle_by_partition:
    enabled: true

Partition Fields

fields (optional, array) - List of partition field definitions. Each field must reference a column that exists in the schema - Ingest Engine validates this at pipeline startup and fails fast if a field is missing. Each entry has:
- name - Column name in the schema to partition by.
- transform (default: identity) - How to derive the partition value from the column. Available transforms:
  - identity - Use the field value as-is. Works with any column type. Good for low-cardinality dimensions like region, event_type, or date columns.
  - hour - Extract the hour bucket from a timestamp column. Requires UnixMicros, UnixMillis, UnixNanos, or DateTime type. Creates hourly partition directories.
  - month - Extract the month from a timestamp column (same type restrictions as hour).
  - year - Extract the year from a timestamp column (same type restrictions as hour).
Partition directory paths are formatted as field=value and nested for multiple fields - for example, country=US/ts=488689 for a two-field partition spec.

Time Pattern

time_pattern (optional, string) - A strftime-compatible format pattern (e.g., %Y/%m/%d or year=%Y/month=%m/day=%d). When set, Ingest Engine adds a watermark-based rolling policy: once the pipeline watermark advances past the current partition’s time boundary (determined by formatting both the watermark and the file’s representative timestamp with this pattern and comparing the strings), the file is closed and a new one is started. This is independent of the fields-based partitioning and is useful for time-bucketed directory layouts.

Shuffle by Partition

shuffle_by_partition.enabled (default: false) - When enabled and partition fields are defined, Ingest Engine hash-shuffles records by partition key across subtasks before writing. This concentrates all records for a given partition value onto one subtask, which reduces the total number of small files (instead of N subtasks each writing to M partitions, you get roughly M files total). The tradeoff is that skewed partition keys can create hot subtasks. If fields is empty, this setting is ignored.

File Naming

Controls how output Parquet files are named within each partition directory (or the table root if there is no partitioning).

file_naming:
  strategy: uuid
  prefix: events
  suffix: parquet

strategy (optional, default: uuid) - How the filename base is generated:
- uuid - Random UUID v4. Default. Guarantees uniqueness across all subtasks.
- uuid_v7 - UUID v7 (time-ordered). Files sort chronologically by creation time.
- ulid - ULID (Universally Unique Lexicographically Sortable Identifier). Like UUID v7 but more compact.
- serial - Sequential {file_index}-{subtask_index} format (e.g., 00001-000). Produces predictable, human-readable names but requires coordination across subtasks.
prefix (optional, string) - Prepended to the filename with a hyphen separator. For example, with prefix: events and strategy: uuid, you get events-550e8400-e29b-41d4-a716-446655440000.parquet.
suffix (optional, string) - File extension. If omitted, Ingest Engine auto-detects from the schema format (typically parquet).

Multipart Upload Tuning

Ingest Engine writes to object storage using multipart uploads. These settings let you tune the upload chunk size and part limits.

multipart:
  target_part_size_bytes: 33554432
  max_parts: 1000

target_part_size_bytes (optional, integer, min: 5242880) - Target size for each part of the multipart upload. Defaults to 32 MB (33554432 bytes). The first part determines the actual part size used for the rest of the file - Ingest Engine measures the first buffered chunk and uses that size going forward. Must be at least 5 MB (the S3 minimum part size).
max_parts (optional, integer) - Maximum number of parts per file. Defaults to 1000. Once this limit is reached, the file is rolled over regardless of rolling policy settings. S3 enforces a hard limit of 10,000 parts per upload; the default of 1000 provides headroom.

Delta Log Commits

When Ingest Engine closes a set of files during a checkpoint, it commits them to the Delta transaction log as Add actions with SaveMode::Append. The table is created automatically on first write if it does not already exist - Ingest Engine sets reader version 3 and writer version 7 on the initial CreateBuilder call. The commit strategy for Delta Lake is PerOperator (not per-subtask), meaning a single coordinator subtask gathers finished files from all subtasks and performs the Delta commit. This ensures atomic commits - either all files from a checkpoint are committed to the log, or none are. During recovery, Ingest Engine checks whether the files from the last checkpoint were already committed to the Delta log (by scanning commits after the last known version). If they were, it skips the duplicate commit and advances the version pointer. This provides exactly-once semantics for Delta log entries.

Complete Example

A Delta Lake sink writing Parquet files to S3, partitioned by event date, with 128 MB file rolling and 5-minute intervals.

table_type:
  type: sink
  sink_config:
    path: s3://my-bucket/delta-tables/events/
    storage_options:
      aws_region: us-east-1
      aws_access_key_id: "${AWS_ACCESS_KEY_ID}"
      aws_secret_access_key: "${AWS_SECRET_ACCESS_KEY}"
    rolling_policy:
      file_size_bytes: 134217728
      interval_seconds: 300
    partitioning:
      fields:
        - name: event_date
          transform: identity
      shuffle_by_partition:
        enabled: true
    file_naming:
      strategy: uuid
      prefix: events
    multipart:
      target_part_size_bytes: 33554432
schema:
  format:
    parquet: {}
  fields:
    - field_name: event_id
      field_type:
        type:
          primitive: Utf8
      nullable: false
    - field_name: event_date
      field_type:
        type:
          primitive: Date32
      nullable: false
    - field_name: payload
      field_type:
        type:
          primitive: Utf8
      nullable: true

See Schema Reference for field type details.

Ingest Engine

Get started

Connectors

Streaming SQL

Operate

Reference

Concepts

Table Path and Storage

Storage Options

Rolling Policy

Partitioning

Partition Fields

Time Pattern

Shuffle by Partition

File Naming

Multipart Upload Tuning

Delta Log Commits

Complete Example

JSON Schema Reference

​Table Path and Storage

​Storage Options

​Rolling Policy

​Partitioning

​Partition Fields

​Time Pattern

​Shuffle by Partition

​File Naming

​Multipart Upload Tuning

​Delta Log Commits

​Complete Example

​JSON Schema Reference

Table Path and Storage

Storage Options

Rolling Policy

Partitioning

Partition Fields

Time Pattern

Shuffle by Partition

File Naming

Multipart Upload Tuning

Delta Log Commits

Complete Example

JSON Schema Reference