AWS Bites Podcast

25. What can you do with Kinesis Data Streams?

Eoin and Luciano are back with some more AWS messaging services material! This time we talk about Kinesis Data Streams and everything there’s to know about them! In this episode of AWS Bites, we cover the difference between data streaming and queue or message buses. How Kinesis can let you process large batches of messages in near real time. Why you might use Kinesis as a queue or pub/sub bus in some cases. A few different ways to use Kinesis with the focus on avoiding too much complexity. We also provide some insights based on our experience with Kinesis Data Streams including real use cases, monitoring tips and some gotchas to be aware of.

In this episode we mentioned the following resources:

Let's talk!

Do you agree with our opinions? Do you have interesting AWS questions you'd like us to chat about? Leave a comment on YouTube or connect with us on:

Luciano
00:03:30

Yeah I always like our e-commerce example so in the context of an e-commerce how can we avail of Kinesis data stream like what kind of utility can we get from from this service and one example that we mentioned before in one of our other episodes was if you need to collect clicks let's say or mouse movements from the users in the e-commerce because that will give us an information what kind of products they're possibly interested in what we could do is that we can collect this data almost in real time from the UI send it to a Kinesis data stream and then we can have some processing logic that it's getting the data in real time figuring out which products maybe are relevant for the users and we can send this data back maybe to a web socket or some other API and refresh the UI in real time with interesting suggestions so that's a cool use case and we could definitely implement that with Kinesis data streams. Also another thing that I have done in the past is basically you can capture logs for instance from CloudWatch logs or other sources and you can store them in Kinesis data streams and then you can let the Lambda integration process these logs pretty much in real time and you can do all sorts of things with that. One use case I've done before is basically if you store things that allow you to extrapolate metrics then from your Lambda you can actually ingest this information as CloudWatch metrics so you can basically transform logs into metrics by using Kinesis data stream as a quick way to transport the data and then start some processing. Any other example comes to mind on your side?

Eoin
00:05:22

Yeah, speaking of metrics there are quite often a few cases where AWS, CloudWatch metrics for some services they're missing some of the metrics you want. Often for things like when you're trying to track the number of resources for pricing you know you want to be alerted when you're using too many containers or number of vc2 instances you can use the last time we were talking about event bridge events that come from AWS out of the box so you'll get state transition events when a container starts or stops similar with EC2 instances. With event bridge it integrates into Kinesis data streams quite easily so you can set a data stream as your target for an event bridge rule you can take all of those events that are coming relating to ECS containers and then you can process them in in batches in lambda or another consumer and use it to figure out okay at any given point in time how many containers are running how many containers are running in each state what is the duration of a running container and you can create your own custom metrics then. This is something I've done a few times with things like ECS and AWS batch where the metrics out of the box have a few gaps and it works pretty well. I suppose you also see other interesting use cases like people are building a delivery service you know like uber deliveroo that kind of service if you can imagine an architecture for that system you've got a lot of real-time event processing so you're collecting orders you're collecting responses from drivers all of those cases are things where you would say okay this is a delivery stream I could use Kinesis or Kafka for this. Yeah absolutely. So I think that's a good thing. Yeah absolutely.

Luciano
00:07:07

Yeah we should maybe talk now about some of the main features of Kinesis Data Stream. So we said that the data is going to stay there but of course it's not going to stay there indefinitely you can configure a retention time by default is 24 hours but you can scale that up to one year so you can keep your data around for a very long time if you want to. Another thing that is worth mentioning is that you get very very good latency it's under a second to basically when you store that first record from when you actually consume it and can be around a second or 70 milliseconds depending on how you use it so we will talk more about this later on. And then messages in Kinesis are called records as you might have guessed at this point but a record generally has a sequence number and a partition key and then there is of course a date of art so the payload itself which is the part you are interested in processing. Another thing that we mentioned to some extent but is worth highlighting a little bit more in detail is the concept of shards. Shards are the unit of scale for Kinesis and I suppose you can imagine shards as different locations where the data is being distributed and partition keys are basically is the concept used to select given a record in which shard is going to end up so this is very similar to many other distributed systems so it's not magic in that sense it uses the same principles as many other distributed systems. And of course that gives us other guarantees so you're not going to have absolute ordering of messages but you will have relative ordering per shard so only the data ending up in the same shard will respect partial ordering there but if you look if you fetch data from different shards they will not necessarily be in order.

Eoin
00:09:09

Exactly that's a little bit like what we talked about SQS and message group IDs it's a similar principle yeah and of course what that means is that with guaranteed ordering you always have physics getting in the way and putting limits on you so we should talk about those limits because the limits of Kinesis and what you can do with a single shard really dictate the architecture of your system so normally we talk about quotas and limits at the end but we're going to talk about this a little bit earlier because you get very fixed capacity and you have to think about that and how it impacts the architecture the scalability and the level of concurrency. It makes your architecture a little bit more complex and it kind of makes you realize that streaming data applications are inherently more complex than queues and pubsub so it's probably worth saying at this point if you don't need any of the specific characteristics of these services it's always better to stick with something simple like SQS and SNF Enbridge because they do give you you can get low latency and scale and massive concurrency with those systems you don't need a streaming data system to get that it's only if you really have specific needs that some of the Kinesis or Kafka features will give you just around the number of messages you can get in a batch for example or the integrations with specific other systems those are the only reasons where you really need to reach for those more complex streaming applications but anyway with that out of the way what are these limits so we have strict limits of one megabyte write per shard or a thousand records per second write capacity so that's how much you can write per shard in a second or you can read two megabytes or two thousand records per second from a shard so the right read capacity is double the right capacity and it's very easy to remember because it's one megabyte and one thousand for right but you're limited to how often you

Eoin
00:11:35

at the same time so they they compound you they are not specific for consumers exactly yeah and you also sometimes need to read more than you're writing because you might be catch up you might be reading from older records in the stream so you have to catch up so you can't catch up if the limits would be only the same as the the right limit but there's also the five get you can only call get records you can only have five read transactions per shard per second as well which is another interesting interesting one so you could that means you can read up like with a single read transaction you can actually read 10 000 records so remember we were both talking about this is about batches that gives you an indication right in one read transaction you might be reading 10 000 records or 50 megabytes so you just need to think about that when you're thinking about latency you might say okay i can get like 50 milliseconds i i i've observed that i can get messages in 50 milliseconds so if i just call get records 20 times a second that's fine but you can you can only call it um five times per second so that's really the unit of scale the size you can think about if you if you know what rate you're going to be ingesting at and reading at then you can do the maths and figure figure the rest out as regards other limits soft limits you have a soft limit of either 200 or 500 shards per stream depending on the region but you can always increase that to i think tens of thousands of shards so you can really get high volumes through kinesis and i guess since we're talking about units of scale we know the limits what do you do then how do you scale how does kinesis scale for

Luciano
00:13:16

you yeah that's an interesting one and it's a little bit of a controversial topic because in some cases you will hear kinesis being addressed as a serverless service but i suppose that really depends on the definition of serverless that you have in mind yeah sometimes the argument to that is that it doesn't really scale to zero that means that you have to think about how do you provision kinesis data streams up front and once you do that if you don't use them they don't automatically scale to zero so an interesting news on this field was that in the last reinvent aws announced the on-demand mode for kinesis data stream which helps a little bit to see kinesis data stream in a more serverless way but again they don't scale to zero so you have to worry less about provisioning them in advance but they will never scale to zero and even if you don't use them there is always some cost that you will need to pay anyway so it's interesting like you could argue it's serverless but not quite anyway talking about scalability we at this point we understand that the shard is probably the primary unit to understand how kinesis data stream will scale and the interesting thing is that if you use the provision mode it's up to you to basically do all the calculations up front and allocate the number of shards that you think are going to be enough for you to sustain your read and write throughput you can change that over time like if you realize at some point you need more or you over provision and you want to scale it down you can do that but there are restrictions you can do it only 10 times per day meaning a 24-hour period and then it's interesting that you can only either double or decrease to half the original number of shards funny enough you can do arbitrary amounts if you use the web ui but you can see that the value fluctuates until it actually converts to the value one so there is some magic there if you use the web ui looks like you can go to any arbitrary number but there is still this restriction where you are doubling and decreasing enough until you get to the number you really want to use yeah and

Eoin
00:20:14

especially comparing the on-demand and the provision mode yeah this is this was a surprise because a lot of people have been anticipating an on-demand or kind of serverless version of kinesis for a while and the slight disappointment came from the pricing because it looks like in order to take advantage of on-demand mode it doesn't immediately make it free but it's because it doesn't scale down to zero so the kind of entry level pricing for each if you compare them on demand is actually more expensive than provisioned because with on-demand you pay per stream per hour and with provision you pay per shard per hour so if you if you go with provisioned the lowest provisioned level which is one shard you'll pay about a third almost a quarter of the price of the on-demand baseline which is around four and a half cents per hour because on-demand is charging you per stream it's giving you four shards out of the box but if you don't use any of those shards you're still being charged right so the baseline is higher for on-demand and then with on-demand you're being it is kind of more senseless and serverless in the sense that it's charging you by the data that's flowing through the system the data ingested and the data retrieved whereas with provisioned you're paid by the shard and then the shard gives you a capacity you don't pay as you read so it depends i think the the the if you're trying to judge which is the right mode from a pricing point of view you just have to compare your use case it's going to vary for everybody some cases on demand will be cheaper in other cases provision will be cheaper it depends on your trade-offs between performance scalability cost and the value you're getting out of this

Luciano
00:21:59

feature so it's yeah it's really a business context how frequent do you actually write and read to and from the stream like is it a constant stream or is it something that can be a little bit more spiky yeah yeah all right in terms of observability because we slightly mentioned that before but yeah let's leave let's see maybe what are the most interesting metrics to look after to to see if we are actually using the data streams correctly so of course the the first one and probably the most obvious is the iterator age which can basically tell you the age of let's say that the message the latest messages you still have to process so it's kind of an indicator of how far behind are you in the stream in consuming it so it's a good way to see if your read capacity is is not well tuned as opposed to the rate of producing data so keep that in mind because that might might be very useful for you to to guarantee that you are actually producing and consuming data as in real time as possible another one is throttling and this is a little bit of a controversial topic in my in my opinion because it's you can get accurate throttling metrics because of course you'll get a data point every time you get throttled but if you want to try to anticipate throttling you have to do a little bit of maths on your own and you can look at the get record and put record matrix and the problem with those metrics is that the ones you get by default in cloud watch they are aggregated by minute so if you have very bursty data producing and consuming the data then you don't really know in a minute how much are you doing for instance in a second so you only see the minute value so you might see a value that looks okay but then you see that at the same time you got a throttling for that particular minute so be careful with that and if you really have bursty use cases maybe you want to do something a little bit more custom and maybe record metrics on your own to make sure that you understand if you want to understand in advance when you might incur in throttling otherwise the simple way of looking at this is just look at the matrix for when the throttling actually happens in reading and writing so when you exceed the capacity and then you can adjust your shards or you can adjust your process to avoid the throttling in the future sometimes even a little bit of throttling can be fine because of course the sdk will retry for you so if that's not really slowing down your pipeline probably you don't even need to change the chart so keep a look at this matrix because they will be telling you a lot and you can understand whether you need to change some configuration and how you read and write the data or the number of shards itself At this point maybe we should talk I think you probably have a fair idea on how to use kinesis data stream but this may be good to do a recap and maybe provide a little bit more details right?

Eoin
00:25:13

Yeah okay I'll try there's generally one way I usually try to use kinesis but I'll talk about that last because there are kind of three different ways to use kinesis and I'll talk about that last because there are kind of three different ways to use it there's the api there's the libraries and then you've got lambda and if we talk about the api usage first it probably gives you a sense of how it works under the hood even if you use some of the abstractions so when you're producing messages and sending them into kinesis there's a put records api and it allows you to put 500 records at a given api request and it does that really quickly as well so when you're again when you're comparing to queuing or pubsub kinesis is all about big lots of messages big batches you can put in 500 in a single request you'll remember like with sns and sqs it's 10 with the batch api's so you can put in a megabyte total per record or up to five megabytes for the whole request so there's a bit of maths to be done there as well and when you specify that you put in the partition key which you already mentioned so the partition key is useful for dictating ordering and also the shard it's allocated to so aws will take your partition key produce a hash of it and use that hash to dictate which shard your message goes into because every shard basically has a range of hash keys that it will accept so it's just basically an integer allocation for each shard a range of integers but if you want to override that and you want to be really have lots of fine control over which shard a message goes into you can actually override the key and specify that integer in the message as a decimal value and that's called the explicit hash key generally you don't have to do that but in some cases if you want to be really if you want to control that ordering and the shard allocation yourself you can do that I suppose since you're putting 500 messages as well it's also really important I think a lot of people using Kinesis have been here if you use the put records api you have to ensure that you handle the errors that come back because you can get partial success and the put records api will tell you which ones have failed and then you need to reprocess those so it's not like a you're going to get an exception or just an error code back you have to look at the error response and figure out which ones have succeeded and which ones have failed and redrive so that's the production side the producer but if we look at the consumer side retrieving manually using the api there's quite a lot of complexity to it or at least there's a lot of heavy lifting because you've got multiple shards you can re you can call get records for a single shard so then you need to get first of all find out how many shards there are you establish an integer and then you can you establish an iterator by calling get shard iterator and the iterator is like a pointer that the service is maintaining for you and then with that pointer you call get records but when you establish your connection get an iterator you can specify whether you want to read from the latest message the oldest message in the stream or you can specify a specific point in time or a specific record and then once you have your iterator you call get records in sequence and you can do it up to five times per second and every like we said already i think you can get 10 000 messages in each get records request then you just it's essentially like pagination after that but of course like there's a bit of complexity in that managing multiple shards you might need like multi-threading to do that you have to keep track of where each thread is storing

Eoin
00:29:39

in every single file yeah i mean these systems are ultimately very fully featured distributed logs and i do like that i think that's really good analogy if you think of each shard or partition as just a text file and every record is on a line and then you just need to keep track of the line number and that's like your iterator um the yeah i mentioned that there's complexity here so the kinesis client library is a consumer library written in java that aws provides to make this a bit easier now you do have to understand how it works a little bit it manages that pool of workers and multi-threading for you and it uses dynamo db to track the state um so it's has it can recover and it can keep track of things as shards increase and decrease it does it is written in java it does have some findings for other languages like node js and python that essentially use the java process running as a daemon and use s standard input and standard output to communicate with the node js or the python process so that's the consumer library but there's also a producer library the kpl and that allows you to send messages but it has multi-threading retry logic batching in there for you and it can also try to reduce the chances that you're going to get throttled on you know the number of messages per second limit by packing large numbers of small messages into a single record and that's something that might be complex to implement yourself but if you use the producer library and the consumer library together it happens seamlessly for you so i think especially if you're in a java ecosystem and you're using instances or containers with kinesis those libraries are a good fit but there's still you still have to understand how they work you still have to manage the extra resources so the third way of using kinesis is definitely my favorite and that's with lambda so i i've i've talked enough about the first two do you want to try cover um how lambda works with the

Luciano
00:31:43

event source mapping that we talked about previously sure yeah i'll try to explain how that works but feel free to add more if i'm missing any important either so we already explore something similar when we talk about sqs where basically we say sqs has a bunch of api is for you to fetch the data but then when you use it with lambda it becomes a little bit more magic and you get a lot of stuff magically integrated and working well out of the box and the reason why that happens is because lambda has this component called event source mapping which is basically something that can be used to pull information and trigger a lambda for you so in the case of kinesis this integration exists as well and basically what you can do you can just say use a kinesis stream as a source for a lambda and what happens is that as new data is available in the stream your lambda will be triggered and you will get an event that describes the data in the stream a little bit more involved than that of course because you might want to configure how you get the data how many lambda gets executed concurrently and things like that so of course these integrations don't you can configure and there are specific parameters you can use you can also do things like aggregations i think it's called tumbling window aggregation so you could configure this integration too as it's going through the stream aggregate some data and then trigger your lambda with already a partial aggregated result so you can even let the this event source mapping do some of the work for you and just trigger the lambda with some partial value already calculated other interesting thing is that you could define rules to filter events so you could specify a filter that tells you i'm only interested in triggering the lambda for events that match certain conditions another interesting one is the batch size window count uh count so for instance you could say i want to receive let's say i don't know batches with 30 messages so i want to trigger a lambda when i accumulate 30 messages but you can also say unless maybe a certain amount of time has passed so i want to receive a smaller batch if i'm waiting too long so you could configure this batch size and window to say what's the maximum amount of time you want to wait or if you can produce batches of a certain size within that unit start to trigger the lambda anyway another interesting one is the parallelization factor which is what is going to tell us how many lambdas we are going to potentially spin up concurrently for processing that particular data stream and basically what what you can say you specify this value and that value gives you the number of lambda concurrent repair shard so let's make an example if you have two shards and you say that parallelization factor is four

Luciano
00:37:41

what those are and what they can give you yeah so you have kinesis data analytics which is if you ever use fling apache fling is something very close to that probably implements the same api or probably is using yeah but it's basically you can define processing logic and aggregation in real time directly on the stream so you basically define yeah how to get the data aggregated and produce new data so it's kind of that analytics use case is the best use case for seeing this particular kinesis variation the other one is kinesis pharaohs which is also quite common and the idea is that sometimes you just want to flush all the data in a stream somewhere so rather than writing all that integration yourself you can use pharaohs and integrates automatically with things like s3 redshift elastic search or even apis through api gateway or directly calling http endpoints so that's a good way when you just want to store the data or propagate it somewhere else you just use pharaohs write it to the stream and let pharaohs do the rest of the integration you can also use that by the way to connect different kinesis data streams together because you could use pharaohs to say move the data from one stream to another so the destination on pharaohs can be another kinesis data stream there is another one i think related to videos you can even have kinesis video streams but i will leave that aside for this episode because it's a little bit of a snowflake i think it's the naming is confusing in reality it's in the family

Help us to make this transcription better! If you find an error, please submit a PR with your corrections.