> ## Documentation Index > Fetch the complete documentation index at: https://docs.e6data.com/llms.txt > Use this file to discover all available pages before exploring further. # 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. ```yaml theme={"theme":{"light":"github-light","dark":"github-dark"}} 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:** ```yaml theme={"theme":{"light":"github-light","dark":"github-dark"}} 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:** ```yaml theme={"theme":{"light":"github-light","dark":"github-dark"}} storage_options: google_service_account_key: # OR google_service_account_path: /path/to/credentials.json ``` **Azure Blob Storage:** ```yaml theme={"theme":{"light":"github-light","dark":"github-dark"}} 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](https://docs.rs/object_store/latest/object_store/) 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. ```yaml theme={"theme":{"light":"github-light","dark":"github-dark"}} 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`. ```yaml theme={"theme":{"light":"github-light","dark":"github-dark"}} 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). ```yaml theme={"theme":{"light":"github-light","dark":"github-dark"}} 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. ```yaml theme={"theme":{"light":"github-light","dark":"github-dark"}} 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. ```yaml theme={"theme":{"light":"github-light","dark":"github-dark"}} 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](/ingest/reference/schema) for field type details. *** ## JSON Schema Reference
Table Schema ```json theme={"theme":{"light":"github-light","dark":"github-dark"}} { "type": "object", "required": ["table_type"], "properties": { "table_type": { "type": "object", "required": ["type", "sink_config"], "properties": { "type": { "const": "sink" }, "sink_config": { "type": "object", "required": ["path"], "properties": { "path": { "type": "string", "description": "URI of the Delta Lake table (s3://, gs://, az://, or local path)" }, "storage_options": { "type": "object", "additionalProperties": { "type": "string" }, "description": "Cloud storage credentials and options passed to the object store client" }, "rolling_policy": { "type": "object", "properties": { "file_size_bytes": { "type": "integer", "description": "Roll after reaching this size in bytes" }, "interval_seconds": { "type": "integer", "minimum": 1, "description": "Roll after this many seconds since file open" }, "inactivity_seconds": { "type": "integer", "minimum": 1, "description": "Roll after this many seconds of no writes" } } }, "partitioning": { "type": "object", "properties": { "time_pattern": { "type": "string", "description": "strftime pattern for watermark-based rolling" }, "fields": { "type": "array", "items": { "type": "object", "required": ["name"], "properties": { "name": { "type": "string" }, "transform": { "type": "string", "enum": ["identity", "hour", "year", "month"], "default": "identity" } } } }, "shuffle_by_partition": { "type": "object", "properties": { "enabled": { "type": "boolean", "default": false } } } } }, "file_naming": { "type": "object", "properties": { "strategy": { "type": "string", "enum": ["serial", "uuid", "uuid_v7", "ulid"], "default": "uuid" }, "prefix": { "type": "string" }, "suffix": { "type": "string" } } }, "multipart": { "type": "object", "properties": { "target_part_size_bytes": { "type": "integer", "minimum": 5242880, "description": "Target size for each multipart upload part" }, "max_parts": { "type": "integer", "description": "Maximum number of parts per multipart upload" } } } } } } } } } ```
This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.