Optimize write throughput for Amazon Kinesis Information Streams

Amazon Kinesis Information Streams is utilized by many shoppers to seize, course of, and retailer information streams at any scale. This degree 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 software is accumulating clickstream information from an internet software or recording telemetry information from billions of Web of Issues (IoT) gadgets, streaming purposes are extremely vulnerable to a various quantity of knowledge ingestion. Typically such a big and sudden quantity of knowledge could possibly be the factor we least anticipate. As an example, take into account software logic with a retry mechanism when writing information to a Kinesis information stream. In case of a community failure, it’s frequent to buffer information domestically and write them when connectivity is restored. Relying on the speed that information is buffered and the period of connectivity problem, the native buffer can accumulate sufficient information that would saturate the out there write throughput quota of a Kinesis information stream.

When an software makes an attempt to write down extra information than what’s allowed, it can obtain write throughput exceeded errors. In some situations, not having the ability to handle these errors in a well timed method may end up in 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 establish 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 might be precious in addressing these errors.

Why can we get write throughput exceeded errors?

Write throughput exceeded errors are typically brought on by three completely different situations:

• The best is the case the place the producer software is producing extra information than the throughput out there within the Kinesis information stream (the sum of all shards).
• Subsequent, we now have the case the place information distribution isn’t even throughout all shards, generally known as scorching shard problem.
• Write all through errors will also be brought on by an software selecting a partition key to write down information at a charge exceeding the throughput provided by a single shard. This case is considerably just like scorching shard problem, however as we see later on this publish, not like a scorching shard problem, you’ll be able to’t remedy this drawback by including extra shards to the info stream. This habits is usually generally known as a scorching key problem.

Earlier than we talk about methods 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. When you view the main points of an information stream utilizing the describe-stream operation within the AWS Command Line Interface (AWS CLI), you’ll be able to truly see this key vary task:

$ 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 software invokes the PutRecord or PutRecords API, the service calculates a MD5 hash for the PartitionKey specified within the file. The ensuing hash is used to find out which shard to retailer that file. You may take extra management over this course of by setting the ExplicitHashKey property within the PutRecord request to a hash key that falls inside a selected shard’s key vary. As an example, setting ExplicitHashKey to 0 will assure that file is written to shard ID shardId-0 within the stream described within the previous code snippet.

How partition keys are distributed throughout out there shards performs an important position in maximizing the out there throughput in a Kinesis information stream. When the partition key getting used is repeated regularly in a approach 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 could 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 file. Let’s say that you’ve 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 one in every of these URLs went viral and it began receiving 1,000 requests per second. Though the scenario is optimistic from a enterprise viewpoint, you’re now on the point of 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 based mostly on the requests per second quota as a result of even with elevated requests, you’re nonetheless producing roughly 11 KB of knowledge.

You may remedy this drawback both through the use of a UUID for every request because the partition key so that you simply share the overall 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 eat information. Altering the partition key to a UUID can be problematic in order for you efficiency metrics from a given URL to be at all times processed by the identical shopper occasion or if you wish to preserve the order of information on a per-URL foundation.

Figuring out the precise explanation for write all through exceeded errors is a vital step in remediating them. Within the subsequent sections, we talk about methods to establish the foundation trigger and remediate this drawback.

Figuring out the reason for write throughput exceeded errors

Step one in fixing an issue is that understanding that it exists. You should use the WriteProvisionedThrougputExceeded metric in Amazon CloudWatch on this case. You may correlate the spikes within the WriteProvisionedThrougputExceeded metric to the IncomingBytes and IncomingRecords metrics to establish whether or not an software is getting throttled because of the measurement of knowledge or the variety of information written.

Let’s take a look at a number of assessments we carried out in a stream with two shards for example numerous situations. On this occasion, with two shards in our stream, complete throughput out there to our producer software 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’ll be able to see within the graph on the left of the next determine, our WriteProvisionedThroughputExceedded errors depend went up. The graph on the proper reveals 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 nicely 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 out there within the stream or if we’re having a scorching shard problem 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 assessments, we enabled enhanced shard-level metrics, and we will see within the following determine that each shards equally reached their quota in our first check.

Now we have seen Kinesis information streams containing 1000’s of shards harnessing the ability of infinite scale in Kinesis information streams. Nevertheless, plotting enhanced shard-level metrics on a such massive stream might not present a straightforward to technique 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 gadgets, 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 usually are not enabled by default. When 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’ll be able to solely question the latest 3 hours of knowledge. Enhanced shard-level metrics additionally incur further prices for CloudWatch metrics and it could be price prohibitive to have it at all times on in information streams with a whole lot of shards.

One attention-grabbing situation is when the workload is bursty, which may make the ensuing CloudWatch metrics graphs quite baffling. It’s because Kinesis publishes CloudWatch metric information aggregated at 1-minute intervals. Consequently, though you’ll be able to see write throughput exceeded errors, your incoming bytes/information graphs could also be nonetheless inside the limits. As an example this situation, we modified our check to create a burst of writes exceeding the boundaries 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 proper appear superb.

To reinforce the method of figuring out write throughput exceeded errors, we developed a CLI instrument referred to as Kinesis Sizzling Shard Advisor (KHS). With KHS, you’ll be able to view shard utilization when shard-level metrics usually are not enabled. That is notably helpful for investigating a problem retrospectively. It will probably additionally present most regularly written keys to a specific shard. KHS reviews shard utilization by studying information and aggregating them per second intervals based mostly on the ApproximateArrivalTimestamp within the file. Due to this, you may as well perceive shard utilization drivers throughout bursty write workloads.

By working the next command, we will get KHS to examine the info 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 reveals the utmost bytes per second charge noticed, complete bytes ingested, most information per second noticed, and the overall 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 just like the graph you get in CloudWatch metrics, besides the KHS graph reviews on a per-second foundation. As an example, within the following determine, we will see how the producer was going by means of a sequence of bursty writes adopted by a throttling occasion throughout our check case.

Operating the identical command with the -aggregate-key possibility allows partition key distribution evaluation. It generates a further graph for every shard displaying the important thing distribution, as proven within the following determine. For our check situation, we will solely see every key getting used one time as a result of we used a brand new UUID for every file.

As a result of KHS reviews based mostly on information saved in streams, it creates an enhanced fan-out shopper at startup to stop utilizing the learn throughput quota out there 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 whole lot of information throughout evaluation. As an example, assume you could have a stream with 100 shards. If all of them are absolutely utilized throughout a minute window specified utilizing -from and -to arguments, the host working KHS will obtain a minimum of 1 MB * 100 * 60 = 6000 MB = roughly 6 GB information. To keep away from this type of extreme information switch and velocity up the evaluation course of, we suggest first utilizing the WriteProvisionedThroughoutExceeded CloudWatch metric to establish a time interval while you skilled throttling and use a small window (equivalent to 10 seconds) with KHS. You may also 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 (equivalent to a hash map storing the partition key and depend) for partition key distribution evaluation may simply exhaust the out there 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 occasions a key has been used. Consequently, the quantity you see within the report needs to be taken as an approximate worth quite 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 methods to establish them, let’s take a look at methods to cope with 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. Contemplate our earlier pattern software logging efficiency metrics information for every internet request served by a fleet of internet servers. When implementing this retry mechanism, it is best to keep in mind that information that aren’t written to the Kinesis stream are going to be in host system’s reminiscence. The primary problem with that is, if the host crashes earlier than the information could possibly be written, you’ll expertise information loss. Eventualities equivalent to monitoring internet request efficiency information could be extra forgiving for such a information loss than situations like monetary transactions. You need to consider sturdiness ensures required to your software and make use of strategies to attain them.

The second problem is that information ready to be written to the Kinesis information stream are going to eat the host system’s reminiscence. While 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 technique to keep away from exhausting the host system’s reminiscence.

With the suitable retry logic in place, if you happen to obtain write throughput exceeded errors, you should use the strategies we mentioned earlier to establish the trigger. After you establish the foundation trigger, you’ll be able to select the suitable remediation step:

• If the producer software is exceeding the general stream’s throughput, you’ll be able to add extra shards to the stream to extend its write throughput capability. When including shards, the Kinesis information stream makes the brand new shards out there 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 by means of the AWS SDK, or the ShardCount property within the AWS CloudFormation template used to create the stream.
• If the producer software is exceeding the throughput restrict of some shards (scorching shard problem), choose one of many following choices based mostly on shopper necessities:
  • If locality of knowledge is required (information with the identical partition key are at all times processed by the identical shopper) or an order based mostly 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 based mostly on the present partition key isn’t required, change the partition key scheme to extend its distribution.
• If the producer software is exceeding the throughput restrict of a shard as a result of a single partition key (scorching key problem), change the partition key scheme to extend its distribution.

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

Conclusion

You need to now have a stable understanding of the frequent causes of write throughput exceeded errors in Kinesis information streams, methods to diagnose them, and what actions to take to appropriately cope with them. We hope that this publish will allow you to make your Kinesis Information Streams purposes extra sturdy. If you’re simply beginning with Kinesis Information Streams, we suggest referring to the Developer Information.

When you’ve got any questions or suggestions, please go away them within the feedback part.

Concerning the Authors

Buddhike de Silva is a Senior Specialist Options Architect at Amazon Net Companies. Buddhike helps clients run massive 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 Companies. He’s keen about growing intuitive product experiences that remedy complicated buyer issues and allow clients to attain their enterprise targets.