Skip to main content
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. Ingest 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.

​
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. Ingest Engine uses it to deserialize incoming data into typed Arrow columns. See the Schema Reference 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 Ingest 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, Ingest 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

Ingest 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.

​
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:
Sink - write processed events with small batches for low latency:

​
JSON Schema Reference

Last modified on October 1, 2026