> 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/message-queues/kinesis.md). # Kinesis **Source** | **Sink** The Kinesis connector lets you read from and write to Amazon Kinesis Data Streams. It uses the AWS SDK directly - not the Kinesis Client Library (KCL) - so there is no DynamoDB lease table or external coordination. e6 Ingestion Engine manages shard assignment, offset tracking, and checkpointing internally. Authentication is handled through the standard AWS credential chain (environment variables, instance profiles, IRSA). There is no connection profile to configure - credentials come from the environment, so every Kinesis table definition is self-contained. *** ## Reading from Kinesis To consume records from a stream, configure a source with the stream name, an offset mode, and a schema describing the record structure. ```yaml stream_name: user-events aws_region: us-east-1 type: offset: latest schema: format: json: {} fields: - field_name: user_id field_type: type: primitive: Int64 nullable: false - field_name: event_type field_type: type: primitive: Utf8 nullable: false ``` ### Stream and Schema * **`stream_name`** (required) - The name of the Kinesis stream to consume from. * **`aws_region`** (optional) - AWS region where the stream lives (e.g., `us-east-1`, `eu-west-1`). If omitted, the region is resolved from the environment - the `AWS_REGION` or `AWS_DEFAULT_REGION` environment variable, the `~/.aws/config` file, or the EC2/ECS instance metadata service. The `schema` block defines the expected structure of the records. e6 Ingestion Engine uses it to deserialize incoming data into typed Arrow columns. See the [Schema Reference](/ingestion-engine/get-started/schema.md) for field type details. The `format` field inside `schema` determines how records are deserialized: * **`json: {}`** - JSON records (most common) * **`raw_bytes: {}`** - Pass through as raw bytes ### Where to Start Reading The `offset` field inside `type` determines where e6 Ingestion Engine begins consuming when a pipeline starts for the first time: * **`latest`** - Start from the tip of each shard. Skip everything already in the stream. Use this when you only care about new data going forward. Maps to the Kinesis `LATEST` shard iterator type. * **`earliest`** - Start from the oldest available records (the trim horizon). Replay the entire stream from the beginning. Use this for backfills or when you need complete history. Maps to the Kinesis `TRIM_HORIZON` shard iterator type. **Important:** The `offset` setting only applies on the very first start of a pipeline, or when there is no checkpoint to restore from. After that, e6 Ingestion Engine tracks offsets through its own checkpoint system using Kinesis sequence numbers. On recovery from a failure, it always resumes from the last successful checkpoint - the `offset` setting is ignored. If new shards appear (from a resharding operation) that were not known at checkpoint time, those new shards start from the configured `offset`. ### Shard Assignment e6 Ingestion Engine distributes shards across parallel subtasks using consistent hashing on the shard ID. Each subtask only reads shards whose hash maps to its index (`shard_hash % parallelism == task_index`). The source polls for new shards every second, so resharding events (splits and merges) are picked up automatically without restarting the pipeline. ### Throttling and Retries If the Kinesis API returns a `ProvisionedThroughputExceededException` or `KMSThrottlingException`, the source retries with exponential backoff (starting at 400ms, doubling each attempt) up to 5 times before failing. Expired shard iterators are handled transparently - the source requests a new iterator and continues reading. *** ## Writing to Kinesis To produce records to a stream, configure a sink with the stream name and optional batching parameters. ```yaml stream_name: processed-events aws_region: us-east-1 type: records_per_batch: 100 batch_flush_interval_millis: 1000 schema: format: json: {} fields: - field_name: user_id field_type: type: primitive: Int64 nullable: false - field_name: event_type field_type: type: primitive: Utf8 nullable: false ``` ### Stream * **`stream_name`** (required) - The name of the Kinesis stream to write to. * **`aws_region`** (optional) - AWS region. Same resolution behavior as the source. ### Batching The sink buffers records and writes them in batches using the Kinesis `PutRecords` API. A batch is flushed when any of these conditions is met: * **`records_per_batch`** (optional, default `500`) - Maximum number of records to buffer before flushing. The Kinesis `PutRecords` API has a hard limit of 500 records per call, so this is also the maximum allowed value. * **`batch_max_buffer_size`** (optional, default `4000000`) - Maximum total size of buffered records in bytes before flushing. The Kinesis `PutRecords` API has a hard limit of 5 MB per call; the default of \~4 MB leaves headroom for partition keys and overhead. * **`batch_flush_interval_millis`** (optional, default `1000`) - Maximum time in milliseconds to hold records before flushing, regardless of batch size. Prevents high-latency delivery during low-throughput periods. ### Partition Keys Each record is assigned a random UUID as its partition key. This distributes records evenly across shards, maximizing write throughput. If you need records for the same entity to land in the same shard, consider using a Kafka-style key field in a future version. ### Delivery Guarantee The sink provides at-least-once delivery. On each checkpoint, all buffered records are flushed to Kinesis with retries (up to 30 attempts with exponential backoff from 100ms to 10 seconds). If a `PutRecords` call partially fails, only the failed records are retried. Between checkpoints, batches are flushed with up to 20 retry attempts. *** ## Authentication Kinesis does not use a connection profile. Credentials are resolved from the standard AWS credential chain, in order: 1. **Environment variables** - `AWS_ACCESS_KEY_ID` and `AWS_SECRET_ACCESS_KEY` (and optionally `AWS_SESSION_TOKEN` for temporary credentials) 2. **Shared credentials file** - `~/.aws/credentials` 3. **IAM instance profile** - EC2 or ECS task role 4. **IAM Roles for Service Accounts (IRSA)** - EKS pod identity via the `AWS_WEB_IDENTITY_TOKEN_FILE` environment variable The IAM principal must have permissions for `kinesis:GetShardIterator`, `kinesis:GetRecords`, `kinesis:ListShards` (source), and `kinesis:PutRecords` (sink) on the target stream. *** ## Complete Example A full configuration reading JSON events from a Kinesis stream and writing processed results back to a different stream. **Source** - read user events from the tip of the stream: ```yaml stream_name: user-events aws_region: us-east-1 type: offset: latest schema: format: json: {} fields: - field_name: user_id field_type: type: primitive: Int64 nullable: false - field_name: event_type field_type: type: primitive: Utf8 nullable: false - field_name: timestamp field_type: type: primitive: TimestampNanosecond nullable: false - field_name: payload field_type: type: primitive: Utf8 nullable: true ``` **Sink** - write processed events with small batches for low latency: ```yaml stream_name: processed-events aws_region: us-east-1 type: records_per_batch: 100 batch_max_buffer_size: 1000000 batch_flush_interval_millis: 500 schema: format: json: {} fields: - field_name: user_id field_type: type: primitive: Utf8 nullable: false - field_name: event_type field_type: type: primitive: Utf8 nullable: false - field_name: timestamp field_type: type: primitive: TimestampNanosecond nullable: false ``` *** ## JSON Schema Reference
Table Schema ```json { "type": "object", "required": ["stream_name", "type"], "properties": { "stream_name": { "type": "string", "description": "Kinesis stream name" }, "aws_region": { "type": "string", "description": "AWS region (resolved from environment if omitted)" }, "type": { "description": "Discriminated by presence of 'offset' (source) or batching fields (sink)", "oneOf": [ { "title": "Source", "type": "object", "required": ["offset"], "properties": { "offset": { "type": "string", "enum": ["latest", "earliest"] } } }, { "title": "Sink", "type": "object", "properties": { "records_per_batch": { "type": "integer", "default": 500, "maximum": 500, "description": "Max records per PutRecords call" }, "batch_max_buffer_size": { "type": "integer", "default": 4000000, "description": "Max batch size in bytes" }, "batch_flush_interval_millis": { "type": "integer", "default": 1000, "description": "Max time (ms) before flushing" } } } ] } } } ```
--- # 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/message-queues/kinesis.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.