> For the complete documentation index, see [llms.txt](https://docs.e6data.com/ingestion-engine/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.e6data.com/ingestion-engine/connectors/lakehouse/delta.md). # 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. e6 Ingestion Engine parses the URI scheme to determine the storage backend - `s3://`, `gs://`, `az://` (or `abfs://`), or a local filesystem path. ```yaml 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. e6 Ingestion 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, e6 Ingestion Engine falls back to its built-in credential provider (IAM roles, environment variables, instance metadata). **AWS S3:** ```yaml 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, e6 Ingestion 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, e6 Ingestion Engine automatically enables path-style access and allows HTTP (useful for MinIO or LocalStack). **Google Cloud Storage:** ```yaml storage_options: google_service_account_key: # OR google_service_account_path: /path/to/credentials.json ``` **Azure Blob Storage:** ```yaml 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 ``` {% hint style="success" %} 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. {% endhint %} *** ## Rolling Policy The rolling policy controls when e6 Ingestion 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 rolling_policy: file_size_bytes: 134217728 interval_seconds: 300 ``` * **`file_size_bytes`** (optional, integer) - Roll after the file reaches this many bytes. e6 Ingestion 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, e6 Ingestion 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), e6 Ingestion 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. e6 Ingestion Engine uses Iceberg-style partition transforms internally - each record is evaluated against the partition fields and routed to a directory named `field=value`. ```yaml 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 - e6 Ingestion 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, e6 Ingestion 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, e6 Ingestion 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 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, e6 Ingestion Engine auto-detects from the schema format (typically `parquet`). *** ## Multipart Upload Tuning e6 Ingestion Engine writes to object storage using multipart uploads. These settings let you tune the upload chunk size and part limits. ```yaml 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 - e6 Ingestion 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 e6 Ingestion 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 - e6 Ingestion 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, e6 Ingestion 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 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](/ingestion-engine/get-started/schema.md) for field type details. *** ## JSON Schema Reference
Table Schema ```json { "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" } } } } } } } } } ```
--- # Agent Instructions This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com. ## Querying This Documentation If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question. Perform an HTTP GET request on the current page URL with the `ask` query parameter, and the optional `goal` query parameter: ``` GET https://docs.e6data.com/ingestion-engine/connectors/lakehouse/delta.md?ask=&goal= ``` `ask` is the immediate question: it should be specific, self-contained, and written in natural language. `goal` is optional and describes the broader end goal you are ultimately trying to accomplish on behalf of the user. GitBook uses it to tailor the answer towards what is most useful for that goal. The response will contain a direct answer to the question and relevant excerpts and sources from the documentation. Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.