Optimize write throughput for Amazon Kinesis Knowledge Streams

Amazon Kinesis Knowledge Streams is utilized by many purchasers to seize, course of, and retailer information streams at any scale. This stage of unparalleled scale is enabled by dividing every information stream into a number of shards. Every shard in a stream has a 1 Mbps or 1,000 information per second write throughput restrict. Whether or not your information streaming utility is accumulating clickstream information from an internet utility or recording telemetry information from billions of Web of Issues (IoT) gadgets, streaming purposes are extremely inclined to a various quantity of information ingestion. Generally such a big and surprising quantity of information may very well be the factor we least anticipate. For example, contemplate utility logic with a retry mechanism when writing information to a Kinesis information stream. In case of a community failure, it’s widespread to buffer information regionally and write them when connectivity is restored. Relying on the speed that information is buffered and the length of connectivity challenge, the native buffer can accumulate sufficient information that might saturate the accessible write throughput quota of a Kinesis information stream.

When an utility makes an attempt to write down extra information than what’s allowed, it should obtain write throughput exceeded errors. In some situations, not having the ability to tackle these errors in a well timed method can lead to information loss, sad clients, and different undesirable outcomes. On this publish, we discover the standard causes behind write throughput exceeded errors, together with strategies to determine them. We then information you on swift responses to those occasions and supply a number of options for mitigation. Lastly, we delve into how on-demand capability mode may be precious in addressing these errors.

Why can we get write throughput exceeded errors?

Write throughput exceeded errors are typically attributable to three totally different situations:

• The best is the case the place the producer utility is producing extra information than the throughput accessible within the Kinesis information stream (the sum of all shards).
• Subsequent, we now have the case the place information distribution is just not even throughout all shards, often known as scorching shard challenge.
• Write all through errors may also be attributable to an utility selecting a partition key to write down information at a charge exceeding the throughput supplied by a single shard. This example is considerably much like scorching shard challenge, however as we see later on this publish, not like a scorching shard challenge, you may’t remedy this drawback by including extra shards to the information stream. This habits is often often known as a scorching key challenge.

Earlier than we focus on learn how to diagnose these points, let’s take a look at how Kinesis information streams arrange information and its relationship to write down throughput exceeded errors.

A Kinesis information stream has a number of shards to retailer information. Every shard is assigned a key vary in 128-bit integer house. Should you view the main points of an information stream utilizing the describe-stream operation within the AWS Command Line Interface (AWS CLI), you may truly see this key vary project:

$ aws kinesis describe-stream --stream-name my-data-stream
"StreamDescription": {
  "Shards": [
    {
      "ShardId": "shardId-000000000000",
      "HashKeyRange": {
        "StartingHashKey": "0",
        "EndingHashKey": 
        "85070591730234615865843651857942052863"
       }
    },
    {
       "ShardId": "shardId-000000000001",
       "HashKeyRange": {
       "StartingHashKey": 
          "85070591730234615865843651857942052864",
       "EndingHashKey": 
         "170141183460469231731687303715884105727"
       }
    }
  ]
}

When a producer utility invokes the PutRecord or PutRecords API, the service calculates a MD5 hash for the PartitionKey specified within the document. The ensuing hash is used to find out which shard to retailer that document. You’ll be able to take extra management over this course of by setting the ExplicitHashKey property within the PutRecord request to a hash key that falls inside a particular shard’s key vary. For example, setting ExplicitHashKey to 0 will assure that document is written to shard ID shardId-0 within the stream described within the previous code snippet.

How partition keys are distributed throughout accessible shards performs a significant function in maximizing the accessible throughput in a Kinesis information stream. When the partition key getting used is repeated regularly in a method that some keys are extra frequent than the others, shards storing these information will probably be utilized extra. We additionally get the identical web impact if we use ExplicitHashKey and our logic for selecting the hash key’s biased in direction of a subset of shards.

Think about you might have a fleet of internet servers logging efficiency metrics for every internet request served right into a Kinesis information stream with two shards and also you used a request URL because the partition key. Every time a request is served, the appliance makes a name to the PutRecord API carrying a 10-bytes document. Let’s say that you’ve got a complete of 10 URLs and every receives 10 requests per second. Below these circumstances, complete throughput required for the workload is 1,000 bytes per second and 100 requests per second. If we assume excellent distribution of 10 URLs throughout the 2 shards, every shard will obtain 500 bytes per second and 50 requests per second.

Now think about considered one of these URLs went viral and it began receiving 1,000 requests per second. Though the state of affairs is optimistic from a enterprise viewpoint, you’re now getting ready to making customers sad. After the web page gained recognition, you’re now counting 1,040 requests per second for the shard storing the favored URL (1000 + 10 * 4). At this level, you’ll obtain write throughput exceeded errors from that shard. You’re throttled primarily based on the requests per second quota as a result of even with elevated requests, you’re nonetheless producing roughly 11 KB of information.

You’ll be able to remedy this drawback both by utilizing a UUID for every request because the partition key so that you simply share the entire load throughout each shards, or by including extra shards to the Kinesis information stream. The strategy you select is determined by the way you wish to devour information. Altering the partition key to a UUID can be problematic if you’d like efficiency metrics from a given URL to be all the time processed by the identical shopper occasion or if you wish to preserve the order of information on a per-URL foundation.

Realizing the precise reason for write all through exceeded errors is a crucial step in remediating them. Within the subsequent sections, we focus on learn how to determine the basis trigger and remediate this drawback.

Figuring out the reason for write throughput exceeded errors

Step one in fixing an issue is that figuring out that it exists. You should use the WriteProvisionedThrougputExceeded metric in Amazon CloudWatch on this case. You’ll be able to correlate the spikes within the WriteProvisionedThrougputExceeded metric to the IncomingBytes and IncomingRecords metrics to determine whether or not an utility is getting throttled as a result of dimension of information or the variety of information written.

Let’s take a look at a number of checks we carried out in a stream with two shards for instance varied situations. On this occasion, with two shards in our stream, complete throughput accessible to our producer utility is both 2 Mbps or 2,000 information per second.

Within the first check, we ran a producer to write down batches of 30 information, every being 100 KB, utilizing the PutRecords API. As you may see within the graph on the left of the next determine, our WriteProvisionedThroughputExceedded errors depend went up. The graph on the appropriate exhibits that we’re reaching the two Mbps restrict, however our incoming information charge is way decrease than the two,000 information per second restrict (Kinesis metrics are revealed at 1-minute intervals, therefore 125.8 and 120,000 as higher limits).

The next figures present how the identical three metrics modified once we modified the producer to write down batches of 500 information, every being 50 bytes, within the second check. This time, we exceeded the two,000 information per second throughput restrict, however our incoming bytes charge is properly beneath the restrict.

Now that we all know that drawback exists, we must always search for clues to see if we’re exceeding the general throughput accessible within the stream or if we’re having a scorching shard challenge as a result of an imbalanced partition key distribution as mentioned earlier. One strategy to that is to make use of enhanced shard-level metrics. Previous to our checks, we enabled enhanced shard-level metrics, and we are able to see within the following determine that each shards equally reached their quota in our first check.

We have now seen Kinesis information streams containing 1000’s of shards harnessing the ability of infinite scale in Kinesis information streams. Nonetheless, plotting enhanced shard-level metrics on a such giant stream could not present a simple to method to discover out which shards are over-utilized. In that occasion, it’s higher to make use of CloudWatch Metrics Insights to run queries to view top-n objects, as proven within the following code (alter the LIMIT 5 clause accordingly):

-- Present high 5 shards with highest incoming bytes
SELECT
SUM(IncomingBytes)
FROM "AWS/Kinesis"
GROUP BY ShardId, StreamName
ORDER BY MAX() DESC
LIMIT 5

-- Present high 5 shards with highest incoming information
SELECT
SUM(IncomingRecords)
FROM "AWS/Kinesis"
GROUP BY ShardId, StreamName
ORDER BY MAX() DESC
LIMIT 5

Enhanced shard-level metrics aren’t enabled by default. Should you didn’t allow them and also you wish to carry out root trigger evaluation after an incident, this selection isn’t very useful. As well as, you may solely question the most up-to-date 3 hours of information. Enhanced shard-level metrics additionally incur extra prices for CloudWatch metrics and it might be value prohibitive to have it all the time on in information streams with a variety of shards.

One attention-grabbing state of affairs is when the workload is bursty, which might make the ensuing CloudWatch metrics graphs relatively baffling. It is because Kinesis publishes CloudWatch metric information aggregated at 1-minute intervals. Consequently, though you may see write throughput exceeded errors, your incoming bytes/information graphs could also be nonetheless throughout the limits. For instance this state of affairs, we modified our check to create a burst of writes exceeding the bounds after which sleep for a number of seconds. Then we repeated this cycle for a number of minutes to yield the graphs within the following determine, which present write throughput exceeded errors on the left, however the IncomingBytes and IncomingRecords graphs on the appropriate appear positive.

To boost the method of figuring out write throughput exceeded errors, we developed a CLI device referred to as Kinesis Sizzling Shard Advisor (KHS). With KHS, you may view shard utilization when shard-level metrics aren’t enabled. That is significantly helpful for investigating a difficulty retrospectively. It may well additionally present most regularly written keys to a selected shard. KHS reviews shard utilization by studying information and aggregating them per second intervals primarily based on the ApproximateArrivalTimestamp within the document. Due to this, you can too perceive shard utilization drivers throughout bursty write workloads.

By working the next command, we are able to get KHS to examine the information that arrived in 1 minute throughout our first check and generate a report:

khs -stream my-data-stream -from "2023-06-22 17:35:00" -to "2023-06-22 17:36:00"

For the given time window, the abstract part within the generated report exhibits the utmost bytes per second charge noticed, complete bytes ingested, most information per second noticed, and the entire variety of information ingested for every shard.

Selecting a shard ID within the first column will show a graph of incoming bytes and information for that shard. That is much like the graph you get in CloudWatch metrics, besides the KHS graph reviews on a per-second foundation. For example, within the following determine, we are able to see how the producer was going via a collection of bursty writes adopted by a throttling occasion throughout our check case.

Operating the identical command with the -aggregate-key possibility permits partition key distribution evaluation. It generates an extra graph for every shard displaying the important thing distribution, as proven within the following determine. For our check state of affairs, we are able to solely see every key getting used one time as a result of we used a brand new UUID for every document.

As a result of KHS reviews primarily based on information saved in streams, it creates an enhanced fan-out shopper at startup to forestall utilizing the learn throughput quota accessible for different shoppers. When the evaluation is full, it deletes that enhanced fan-out shopper.

Due its nature of studying information streams, KHS can switch a variety of information throughout evaluation. For example, assume you might have a stream with 100 shards. If all of them are totally utilized throughout a minute window specified utilizing -from and -to arguments, the host working KHS will obtain no less than 1 MB * 100 * 60 = 6000 MB = roughly 6 GB information. To keep away from this type of extreme information switch and pace up the evaluation course of, we advocate first utilizing the WriteProvisionedThroughoutExceeded CloudWatch metric to determine a time interval if you skilled throttling and use a small window (reminiscent of 10 seconds) with KHS. It’s also possible to run KHS in an Amazon Elastic Compute Cloud (Amazon EC2) occasion in the identical AWS Area as your Kinesis information stream to attenuate community latency throughout reads.

KHS is designed to run in a single machine to diagnose large-scale workloads. Utilizing a naive in-memory-based counting algorithm (reminiscent of a hash map storing the partition key and depend) for partition key distribution evaluation may simply exhaust the accessible reminiscence within the host system. Due to this fact, we use a probabilistic information construction referred to as count-min-sketch to estimate the variety of instances a key has been used. Consequently, the quantity you see within the report ought to be taken as an approximate worth relatively than an absolute worth. In any case, with this report, we simply wish to discover out if there’s an imbalance within the keys written to a shard.

Now that we perceive what causes scorching shards and learn how to determine them, let’s take a look at learn how to take care of this in producer purposes and remediation steps.

Remediation steps

Having producers retry writes is a step in direction of making our producers resilient to write down throughput exceeded errors. Think about our earlier pattern utility logging efficiency metrics information for every internet request served by a fleet of internet servers. When implementing this retry mechanism, it is best to do not forget that information that aren’t written to the Kinesis stream are going to be in host system’s reminiscence. The primary challenge with that is, if the host crashes earlier than the information may very well be written, you’ll expertise information loss. Eventualities reminiscent of monitoring internet request efficiency information could be extra forgiving for the sort of information loss than situations like monetary transactions. It is best to consider sturdiness ensures required in your utility and make use of strategies to attain them.

The second challenge is that information ready to be written to the Kinesis information stream are going to devour the host system’s reminiscence. If you begin getting throttled and have some retry logic in place, it is best to discover that your reminiscence utilization goes up. A retry mechanism ought to have a method to keep away from exhausting the host system’s reminiscence.

With the suitable retry logic in place, for those who obtain write throughput exceeded errors, you should use the strategies we mentioned earlier to determine the trigger. After you determine the basis trigger, you may select the suitable remediation step:

• If the producer utility is exceeding the general stream’s throughput, you may add extra shards to the stream to extend its write throughput capability. When including shards, the Kinesis information stream makes the brand new shards accessible incrementally, minimizing the time that producers expertise write throughput exceeded errors. So as to add shards to a stream, you should use the Kinesis console, the update-shard-count operation within the AWS CLI, the UpdateShardCount API via the AWS SDK, or the ShardCount property within the AWS CloudFormation template used to create the stream.
• If the producer utility is exceeding the throughput restrict of some shards (scorching shard challenge), choose one of many following choices primarily based on shopper necessities:
  • If locality of information is required (information with the identical partition key are all the time processed by the identical shopper) or an order primarily based on partition key’s required, use the split-shard operation within the AWS CLI or the SplitShard API within the AWS SDK to separate these shards.
  • If locality or order primarily based on the present partition key is just not required, change the partition key scheme to extend its distribution.
• If the producer utility is exceeding the throughput restrict of a shard as a result of a single partition key (scorching key challenge), change the partition key scheme to extend its distribution.

Kinesis Knowledge Streams additionally has an on-demand capability mode. In on-demand capability mode, Kinesis Knowledge Streams routinely scales streams when wanted. Moreover, you may change between on-demand and provisioned capability modes with out inflicting an outage. This may very well be significantly helpful if you’re experiencing write throughput exceeded errors however require speedy response to maintain your utility accessible to your customers. In such situations, you may change a provisioned capability mode information stream to an on-demand information stream and let Kinesis Knowledge Streams deal with the required scale appropriately. You’ll be able to then carry out root trigger evaluation within the background and take corrective actions. Lastly, if vital, you may change the capability mode again to provisioned.

Conclusion

It is best to now have a stable understanding of the widespread causes of write throughput exceeded errors in Kinesis information streams, learn how to diagnose them, and what actions to take to appropriately take care of them. We hope that this publish will provide help to make your Kinesis Knowledge Streams purposes extra strong. In case you are simply beginning with Kinesis Knowledge Streams, we advocate referring to the Developer Information.

You probably have any questions or suggestions, please go away them within the feedback part.

In regards to the Authors

Buddhike de Silva is a Senior Specialist Options Architect at Amazon Net Providers. Buddhike helps clients run giant scale streaming analytics workloads on AWS and make one of the best out of their cloud journey.

Nihar Sheth is a Senior Product Supervisor at Amazon Net Providers. He’s enthusiastic about creating intuitive product experiences that remedy complicated buyer issues and allow clients to attain their enterprise objectives.