Connect to Amazon Kinesis

Use Structured Streaming to read and write data to Amazon Kinesis.

Databricks recommends that you enable S3 VPC endpoints so that all S3 traffic is routed on the AWS network.

note

You can reshard with Structured Streaming by increasing the number of shards without interrupting or restarting the stream.

For recommendations on troubleshooting query latency, see Recommendations for reducing latency with Kinesis.

Authentication

Kinesis supports authentication with a Unity Catalog connection, a service credential, or alternate methods such as an instance profile or access keys. See Authentication.

Schema

Kinesis returns records with the following schema:

Column	Type	Description
partitionKey	string	The partition key that identifies which shard the record is assigned to.
data	binary	The record's data blob, as opaque binary.
stream	string	The name or ARN of the Kinesis stream the record was read from.
shardId	string	The ID of the shard the record was read from.
sequenceNumber	string	The unique identifier of the record within its shard.
approximateArrivalTimestamp	timestamp	The approximate time the record was inserted into the stream.

Column	Type	Description
partitionKey	string	The partition key that identifies which shard the record is assigned to.
data	binary	The record's data blob, as opaque binary.
stream	string	The name or ARN of the Kinesis stream the record was read from.
shardId	string	The ID of the shard the record was read from.
sequenceNumber	string	The unique identifier of the record within its shard.
approximateArrivalTimestamp	timestamp	The approximate time the record was inserted into the stream.

To deserialize the data in the data column, cast the field to a string.

Quickstart

The following notebook demonstrates how to run WordCount using Structured Streaming with Kinesis.

Kinesis WordCount with Structured Streaming notebook

Open notebook in new tabOpen in Databricks

Configure Kinesis options

In Databricks Runtime 13.3 LTS and above, you can use Trigger.AvailableNow with Kinesis. See Ingest Kinesis records as an incremental batch.

In Databricks Runtime 16.1 and above, you can use streamARN to identify Kinesis sources. For all Databricks Runtime versions, you must specify either streamName or streamARN, but not both.

warning

Do not switch between streamName and streamARN for an active streaming query. Databricks doesn't support switching between these options mid-stream. Restarting the query can result in duplicate records or data loss. To switch from streamName to streamARN, start a new streaming query with a fresh checkpoint directory.

For the full list of options, see Kinesis.

Add or remove stream sources

In Databricks Runtime 19 and above, you can change the Kinesis source streams for Structured Streaming queries using Spark options, streamName or streamARN.

Add a stream

To add a stream, include it in the streamName or streamARN option list and restart the stream. For each new stream source, the query reads the earliest available offset from the source's shards.

The following example uses the streamName option to add the stream3 Kinesis source to a query that previously read stream1 and stream2:

Python

df = (spark.readStream
  .format("kinesis")
  .option("streamName", "stream1,stream2")
  .load()
)

df.stop()

df = (spark.readStream
  .format("kinesis")
  .option("streamName", "stream1,stream2,stream3")  # Previous value was "stream1,stream2"
  .load()
)

Remove a stream

By default, when you remove a Kinesis stream source from the streamName or streamARN option list, the query fails on restart with a KINESIS_SOURCE_STREAMS_REMOVED_ON_RESTART error. This guarantees that the query doesn't silently skip unread records from the removed stream.

To remove a Kinesis stream source, do the following:

1. Set spark.databricks.kinesis.failOnDataLoss to false in the cluster's Spark configuration and restart the cluster. For more information about failOnDataLoss, see Handle data loss.

2. Remove the stream from the streamName or streamARN option and restart the query. For example, to stop reading stream2:

   Python

   df = (spark.readStream
     .format("kinesis")
     .option("streamName", "stream1,stream2")
     .load()
   )
   
   df.stop()
   
   df = (spark.readStream
     .format("kinesis")
     .option("streamName", "stream1")  # Previous value was "stream1,stream2"
     .load()
   )

You only need to set spark.databricks.kinesis.failOnDataLoss to false and restart the cluster once. After that, removing additional stream sources on that cluster requires only a query restart, not another cluster restart.

note

By default, removing a Kinesis stream source doesn't deregister the enhanced fan-out (EFO) consumer of the stream source, which might continue to incur costs from the cloud provider. To deregister the consumer when the query stops, set the requireConsumerDeregistration option to true. See Kinesis.

To manage consumers directly, see Configure Kinesis enhanced fan-out (EFO) for streaming query reads.

Low latency monitoring and alerting

Alerting use cases require low latency. To minimize latency:

• Verify that your streaming query is the only consumer of the Kinesis stream to optimize fetch performance and avoid Kinesis rate limits.
• Set the option maxFetchDuration to a small value, such as 200ms, to process fetched data as fast as possible. This option is a trade-off: it prioritizes faster processing speed per batch instead of a guarantee that the most recent records are consumed in each batch. For example, if you use Trigger.AvailableNow, a small value might cause your query to lag behind the newest records in the Kinesis stream.
• Set the option minFetchPeriod to 210ms to fetch as frequently as possible.
• In Databricks Runtime 19 and above, set maxPartitions to control the number of Spark tasks assigned to Kinesis. Spark uses min(active shards, maxPartitions) tasks, with each task handling one or more active shards. You cannot set both maxPartitions and shardsPerTask.
• Set the option shardsPerTask or configure the cluster such that # cores in cluster >= 2 * (# Kinesis shards) / shardsPerTask. This guarantees that the background prefetching tasks and the streaming query tasks run concurrently.

Monitor Kinesis metrics

Kinesis reports the number of milliseconds a consumer lags behind the beginning of a stream for each workspace. The avgMsBehindLatest, maxMsBehindLatest, and minMsBehindLatest metrics provide the average, minimum, and maximum milliseconds across all workspaces in the streaming query process. See Monitoring Structured Streaming queries on Databricks.

View source metrics

Use one of the following methods to access source metrics:

Raw Data

If you are running the stream in a notebook, see metrics under the Raw Data tab in the streaming query progress dashboard. The following example shows the source metrics:

JSON

{
  "sources": [
    {
      "description": "KinesisV2[stream]",
      "metrics": {
        "avgMsBehindLatest": "32000.0",
        "maxMsBehindLatest": "32000",
        "minMsBehindLatest": "32000"
      }
    }
  ]
}

Python

After a trigger completes, access the source metrics through lastProgress:

Python

progress = query.lastProgress
print(progress["sources"][0]["metrics"])

Monitor in Amazon CloudWatch

You can also monitor these Amazon CloudWatch metrics:

Metric	Mode	Description
GetRecords.Bytes	Polling	The bytes returned by GetRecords.
GetRecords.Records	Polling	The records returned by GetRecords.
SubscribeToShardEvent.Bytes	EFO	The bytes received from the shard.
SubscribeToShardEvent.Records	EFO	The records received from the shard.
GetRecords.IteratorAgeMilliseconds	Polling	The age of the last record returned. A rising value indicates that the consumer is falling behind.
ReadProvisionedThroughputExceeded	Polling	The number of reads throttled because the shard read capacity was exceeded.

Metric	Mode	Description
GetRecords.Bytes	Polling	The bytes returned by GetRecords.
GetRecords.Records	Polling	The records returned by GetRecords.
SubscribeToShardEvent.Bytes	EFO	The bytes received from the shard.
SubscribeToShardEvent.Records	EFO	The records received from the shard.
GetRecords.IteratorAgeMilliseconds	Polling	The age of the last record returned. A rising value indicates that the consumer is falling behind.
ReadProvisionedThroughputExceeded	Polling	The number of reads throttled because the shard read capacity was exceeded.

Diagnose low throughput

To diagnose low throughput, monitor maxMsBehindLatest and, for polling queries in real-time mode, numAwsRateLimitErrors in the Kinesis source metrics. Also monitor busyTimeFraction in the task utilization metrics. Use the following table to identify the appropriate action.

Metric status	What it means	Resolution
maxMsBehindLatest stays near zero	The query is caught up.	No action needed.
For polling queries in real-time mode, maxMsBehindLatest increases and numAwsRateLimitErrors is repeatedly above zero	AWS throttling limits polling throughput.	Add Kinesis shards or increase minFetchPeriod to reduce polling frequency.
maxMsBehindLatest increases and busyTimeFraction stays near 1	Spark tasks are fully utilized.	Increase compute and the number of tasks.

Metric status	What it means	Resolution
maxMsBehindLatest stays near zero	The query is caught up.	No action needed.
For polling queries in real-time mode, maxMsBehindLatest increases and numAwsRateLimitErrors is repeatedly above zero	AWS throttling limits polling throughput.	Add Kinesis shards or increase minFetchPeriod to reduce polling frequency.
maxMsBehindLatest increases and busyTimeFraction stays near 1	Spark tasks are fully utilized.	Increase compute and the number of tasks.

Avoid slowdowns caused by too many rate limit errors

The connector reduces the amount of data read from Kinesis by half each time it encounters a rate limiting error and records this event in the log with a message: "Hit rate limit. Sleeping for 5 seconds."

You might see these errors while a stream is catching up. If you see these errors after a stream is caught up, you might need to tune the workload by either increasing Kinesis capacity in AWS or by adjusting the prefetching options in Spark.

Each Kinesis shard supports five GetRecords calls per second across polling readers. When N polling queries read the same stream, set minFetchPeriod to at least 200ms * N so their calls to each shard stay within this limit:

Python

.option("minFetchPeriod", "400ms")  # Two polling queries read the same stream.

Ingest Kinesis records as an incremental batch

In Databricks Runtime 13.3 LTS and above, Databricks supports using Trigger.AvailableNow with Kinesis data sources for incremental batch semantics. The following describes the basic configuration:

1. When a micro-batch read triggers in available now mode, the current time is recorded by the Databricks client.
2. Databricks polls the source system for all records with timestamps between this recorded time and the previous checkpoint.
3. Databricks loads these records using Trigger.AvailableNow semantics.

Databricks uses a best-effort mechanism to try and consume all records that exist in Kinesis streams when the streaming query runs. Because of small potential differences in timestamps and a lack of guarantee in ordering in data sources, a triggered batch might not include some records. Omitted records are processed in the next triggered micro-batch.

note

If the query continues failing to fetch records from the Kinesis stream even if there are records, try increasing the maxFetchDuration value.

See AvailableNow: Incremental batch processing.

Handle data loss

warning

Use failOnDataLoss only if your workload can tolerate missing records. Using this incorrectly can result in permanent data loss. If you can't tolerate missing records, restart the stream with a new checkpoint to reprocess all records.

Databricks recommends that you use this only as a temporary mitigation for a data loss issue. Investigate and fix the root cause, such as a Kinesis retention period that is too short.

If a Kinesis shard's records expire before your streaming query reads them, or if you delete and recreate a Kinesis stream with the same name, the query fails with a KINESIS_COULD_NOT_READ_SHARD_UNTIL_END_OFFSET error. See KINESIS_COULD_NOT_READ_SHARD_UNTIL_END_OFFSET.

By default, streaming queries fail when they detect potential data loss. To configure the query to skip unreadable records and continue processing, set spark.databricks.kinesis.failOnDataLoss to false in the cluster's Spark configuration and restart the cluster.

Write to Kinesis

Use the following code snippet as a ForeachSink to write data to Kinesis. It requires a Dataset[(String, Array[Byte])].

note

The following code snippet provides at least once semantics, not exactly once.

Kinesis Foreach Sink notebook

Open notebook in new tabOpen in Databricks

Recommendations for reducing latency with Kinesis

This section has recommendations for troubleshooting various causes of latency for Kinesis streams.

The Kinesis source runs Spark jobs in a background thread to periodically prefetch Kinesis data and then cache the data in Spark executor memory. After each prefetch step completes, the streaming query can process the cached data. The prefetch step significantly affects the observed end-to-end latency and throughput.

Reduce prefetch latency

To optimize for minimal query latency and maximum resource usage, use the following calculation:

total number of CPU cores in the cluster (across all executors) >= total number of Kinesis shards / shardsPerTask.

important

minFetchPeriod can create multiple GetRecords API calls to the Kinesis shard until it reaches ReadProvisionedThroughputExceeded. If an exception occurs, it might not be an issue because the connector maximizes the utilization of the Kinesis shard.

Avoid disk spill

If you have a sudden increase of data volume in your Kinesis streams, the assigned buffer capacity might fill up and not empty fast enough to add new data. Spark spills data from the buffer to disk, which slows down stream processing, and an event appears in the log with a message like the following:

Bash

./log4j.txt:879546:20/03/02 17:15:04 INFO BlockManagerInfo: Updated kinesis_49290928_1_ef24cc00-abda-4acd-bb73-cb135aed175c on disk on 10.0.208.13:43458 (current size: 88.4 MB, original size: 0.0 B)

To avoid spill, increase the cluster memory capacity by adding more nodes or increasing the memory per node, or reduce the configuration parameter fetchBufferSize.

Suspended S3 write tasks

Enable Spark speculation to terminate suspended tasks that would prevent stream processing from proceeding. To ensure that tasks are not terminated too aggressively, tune the quantile and multiplier for this setting carefully. Databricks recommends that you set spark.speculation.multiplier to 3 and spark.speculation.quantile to 0.95 and adjust as needed.

Reduce latency from checkpointing in stateful streams

Databricks recommends using RocksDB with changelog checkpointing for stateful streaming queries. See Enable changelog checkpointing.

Configure Kinesis enhanced fan-out (EFO) for streaming query reads

In Databricks Runtime 11.3 and above, the Databricks Runtime Kinesis connector provides support for using the Amazon Kinesis enhanced fan-out (EFO) feature.

Kinesis enhanced fan-out provides dedicated throughput of 2 MB/s per shard per consumer (maximum of 20 consumers per stream), and delivers records in push mode instead of pull mode.

By default, a Structured Streaming query configured with EFO mode registers itself as a consumer with dedicated throughput and a unique consumer name and consumer ARN (Amazon Resource Name) in Kinesis Data Streams.

By default, Databricks uses the streaming query ID with the databricks_ prefix to name the new consumer. You can optionally specify the consumerNamePrefix or consumerName options to override this behavior. The consumerName must be a string that contains only letters, numbers, and the special characters _ . -.

On query restart, the Kinesis source uses polling mode to replay the latest uncommitted batch if one exists. After the stream replays the uncommitted batch, the source switches back to EFO mode for subsequent reads.

important

A registered EFO consumer incurs additional charges on Amazon Kinesis. To deregister the consumer automatically on query teardown, set the requireConsumerDeregistration option to true. Databricks cannot guarantee de-registration on events such as driver crashes or node failures. In case of job failure, Databricks recommends managing registered consumers directly to prevent excess Kinesis charges.

Offline consumer management using a Databricks notebook

Use the AWSKinesisConsumerManager utility to programmatically register, list, or deregister consumers for Kinesis data streams, instead of manually configuring consumers in your AWS account console. For example, use the utility to create a consumer for a new stream, or, if you plan to permanently stop a stream, use the utility to delete the consumer in AWS.

The consumer manager utility is only available in Scala with compute set to dedicated access mode. See Access modes.

To use this utility in a Databricks notebook:

1. In a new Databricks notebook attached to an active cluster, create an AWSKinesisConsumerManager with required authentication information.

   Scala

   import com.databricks.sql.kinesis.AWSKinesisConsumerManager
   
   val manager = AWSKinesisConsumerManager.newManager()
   .option("serviceCredential", serviceCredentialName)
   .option("region", kinesisRegion)
   .create()

2. List and display consumers.

   Scala

   val consumers = manager.listConsumers("<stream name>")
   display(consumers)

3. Register a consumer for given stream.

   Scala

   val consumerARN = manager.registerConsumer("<stream name>", "<consumer name>")

4. Deregister a consumer for given stream.

   Scala

   manager.deregisterConsumer("<stream name>", "<consumer name>")