Uniform Data Distribution Among Kinesis Data Stream Shards
Tags: kinesis, dataengineering, python3, aws
Published: 2020-11-12 13:55:16

![kinesis](https://dev-to-uploads.s3.amazonaws.com/i/at5s1csj3yyj42duqm3g.png)

# Overview

This post is about how to ***distribute data among kinesis shards uniformly*** but before moving forward go through the ***[kinesis documentation](https://docs.aws.amazon.com/streams/latest/dev/introduction.html)*** to grasp the basic understanding of kinesis data stream. I used Lambda as a consumer for Kinesis data stream, so I will discuss things in Lambda’s context.

Read this **[article](https://read.acloud.guru/deep-dive-into-aws-kinesis-at-scale-2e131ffcfa08)** about Kinesis stream and AWS Lambda.

# Problem
A Kinesis data stream is a set of **[shards](https://docs.aws.amazon.com/streams/latest/dev/key-concepts.html#shard)**. Each shard has a sequence of data records. Each data record has a **[sequence number](https://docs.aws.amazon.com/streams/latest/dev/key-concepts.html#sequence-number)** that is assigned by Kinesis Data Streams.

To put a record in Kinesis stream **Partition Key** must be provided for each record. An MD5 hash function is used to map partition keys to 128-bit integer values and to map associated data records to shards using the hash key ranges of the shards.

![image](https://d2908q01vomqb2.cloudfront.net/b6692ea5df920cad691c20319a6fffd7a4a766b8/2019/12/13/KinesisDataStreamsScaling3_1.png)

The problem arises when the **partition key** is not random and **MD5 hash function** maps the record’s partition key in a specific shard’s hash key range more than the other.


![Alt Text](https://dev-to-uploads.s3.amazonaws.com/i/zxn6qjw55d25opgguwug.png)

In the above image, we can see that **shard#2** has more records than other shards because of the non-random partition key assignment.

Non-uniform distribution of messages causes **Lambda’s Iterator age** to increase and there is a chance to lose data. If we increase the retention period it will increase the operational cost of Kinesis.

```
Iterator age is the time between when the last record in a batch was recorded and when Lambda reads the record. 

Iterator age depends on other parameters, such as Lambda function execution duration, shard count, and batch size. 

For more information, see AWS Lambda CloudWatch Metrics.

Kinesis can retain data up to 7 days and the default retention period is 1 day.
```

# Solution

There are two ways to distribute data among shards uniformly:

1. Assign each record a **unique partition key**, in this way, there is a better chance of uniform distribution.

2. Assign **[ExplicitHashKey](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecordsRequestEntry.html#Streams-Type-PutRecordsRequestEntry-ExplicitHashKey)** to each record. In this way, the **partition key** is not required and we can make sure that the message is dumped in the shard of our choice. In python, we can use [list shards](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/kinesis.html#Kinesis.Client.list_shards) method to get shards information. It will return information in this format:

```json
{  "Shards": [
            {
                "ShardId": "shardId-000000000000",
                "HashKeyRange": {
                    "StartingHashKey": "0",
                    "EndingHashKey": "170141183460469231371588410571"
                },
                "SequenceNumberRange": {
                    "StartingSequenceNumber": "496103516369698317099491240188243335841720557371394"
                }
            },
            {
                "ShardId": "shardId-000000000001",
                "HashKeyRange": {
                    "StartingHashKey": "170141183460469231371588410571",
                    "EndingHashKey": "340282asdasd6337460743176821144"
                },
                "SequenceNumberRange": {
                    "StartingSequenceNumber": "49610351636992132529573630114381723961608490082063351826"
                }
            },
            {
                "ShardId": "shardId-000000000002",
                "HashKeyRange": {
                    "StartingHashKey": "34028236692093846346337460743176821145",
                    "EndingHashKey": "51042355038140769519506191114765231717"
                },
                "SequenceNumberRange": {
                    "StartingSequenceNumber": "49610351637014433274772160737523259679881138443569332258"
                }
            }
        ]
}
```

As we can see that each shard has its own **hash key range**, so by using that information we can generate a key that lies between shard’s ***StartingHashKey*** and ***EndingHashKey***. So by using this approach we can implement a ***round-robin algorithm*** that will put records to each shard one by one for achieving uniform data distribution among Kinesis data stream shards.


# Final Thoughts

The approach discussed in this story is one way to achieve uniform data distribution among Kinesis shards. Please share your feedback about anything that can be improved or I missed. Thank you

## Resources

https://docs.aws.amazon.com/streams/latest/dev/introduction.html