Scale By The Bay 2021: Opher Dubrovsky, Ido Nadler, Scaling your Kafka streaming pipeline....
hi everybody so this talk is about scaling your pipelines can sometimes be a pain but it doesn't always have to be like that i want to start off with a feature of data data is really like cheese if you wait too long your cheese will spoil the same thing with data data usually has a retention period if you wait too long to process it you may end up losing it and if you have an outage and you have a slow recovery this is practically a ticking time bomb so let's talk a bit talk about kafka we love kafka it's in the heart of all our systems but i want to tell you a true story this is a true story the events described in this talk took place at nielsen marketing cloud in 2020 september 24 2020 5 55 pm i get an emergency call from my boss and he says check slack i go to slack and i see mayday mayday earlier today the zoopy keepers crashed and when it start everything started again it caused the kafka instances zero one and zero two to purge their own data please add whoever is relevant this is an outage we had to deal with the consequences of that day were about one day of downtime we had data loss we had five working lost working days uh that were lost uh just to resolve it we had unpleasant customer calls and about 300 in estimates uh in damages that i estimate so in this talk we're going to go over some pros and cons of kafka we'll talk about scaling up scaling down and how you can improve your kafka stream pipeline we'll talk about a few things that could go wrong and then go into some insights and tactics of how to resolve things that hopefully you can take into your own projects before we go on i want to present ourselves my name is offer dubrovsky i'm a director of big data engineering at nielsen and with me is the don nudler a big data team lead we love data we deal with data pipelines kafka serverless and spark and we love it all needs a marketing cloud uh builds marketing audiences and device graphs and that data is very useful for our customers who are marketers for running campaigns and making this business decisions just to explain what audiences are audiences are basically lists of uh user device ids with some label so for example pizza lovers or people who are into running and cycling and each of these groups has an um a number of ids that we have associated with and they are very useful for marketers marketers love these they can slice and dice and intersect those groups and then use them to run marketing campaigns against them so for example if there's some cycling event going on in my area and on my pizzeria chain perhaps i want to run a campaign to all the cycling enthusiasts who are who like pizza and send them some coupon great and useful some technical details about us we're cloud native we run on aws we ingest about 60 terabytes of data and store about a five petabytes in total we're heavy users of spark and of serverless lambda functions in terms of our kafka pipeline we process on a normal day about 25 billion events per day which is about 300 000 events per second but at the peak we processed five millions events per second and in terms if you look at the size of it in terabytes it's about 200 terabytes a day uh 2.5 gigabytes of data per second and peak we had about 5.3 gigabytes per second if you look at our kafka cluster uh it splits throughout 40 different topics and about 9 000 partitions we have 60 brokers in that cluster which equates roughly two instances and the total cost of it is about 540 000 a year in a nutshell i want to say a few words about kafka for people who are less familiar with it so kafka is basically a pub sub messaging system on the left here we have producers of uh that produce data and they send that data in messages to kafka messages are equate equal to events basically and on the right we have consumers that can consume that data kafka is really awesome it's distributed you can add more machines to increase the scale so it's very scalable it's highly available and it's fault tolerant if you dive into the data structure inside uh data is divided into topics topics are just pieces of information that of the same type so we could divide one topic for audiences and another one for data on some machines and so forth and in the topic we can divide it into partitions for convenience of reading this data so the partitions could be of smaller size easier to manage and handle and our consumers basically would read from all of these topics and partitions and this is what makes it useful so i mentioned we love kafka why we love kafka it's just an awesome system it's optimized for high throughput it has data retention built in which means it will delete all data on its own without us having to do anything it supports multiple consumers it acts as a great buffer between the different systems and producers of data and consumers and another nice feature it allows for reprocessing which is very convenient so it's really awesome but there are a few issues with it so one of them is processing bursts the number of consumers that we have to process bursts or to process data at all is limited we can grow the number of consumers but it's limited to the number of partitions so it's a bottleneck so for example in this example in front of me i have three topics i can have one consumer reading from all of them and if i have more data i can scale it up all the way to three consumers but i can't go beyond the number of partitions so we're stuck in this situation with three and if you calculate the throughput that means that the max throughput is basically the number of partitions times the consumer throughput so cons so the max throughput is a glass ceiling that's actually limited by our partitions another issue is over provisioning so um so overall data is is moody and it goes up and down over the day and we really are a consumers and kafka ex for for all purposes are in many cases static that means that we have to over provision it for the times that we have more data and then during those load times we end up paying for resources we don't really need they're not really working very hard during these times and the third issue is over provisioning basically if we need to re if we need longer retention on the data we need more disk space so for example if we have one a high uh throughput holiday season a year we need to provide a provision our cluster for that and then usually what it means is that we're stuck with that higher provisioned cluster that's more expensive moreover there's also the data skew issue so data could be skewed over the partitions meaning not divided equally throughout the partitions and then our consumers reading from the partitions may have different loads of work so some of them may actually have a high throughput and pretty much up to their limits but some of them may be kind of sluggish and idling away wasting our money let's look at the system we used to read from kafka so our old system had spark as a consumer for the different topics we had multiple clusters of spark for the different topics and they were using spark streaming to read data from kafka and store that data in our data lake on s3 the data was coming in as a stream and spark was using windows of six minutes to read the data and and save it during the low hours we had less data so we had we ended up with small files because the data was divided throughout so many partitions each one had just a low amount of data not a lot of files so we ended up with small files if we wanted to correct this in our spark cluster that requires re-partitioning the data and shuffling the data throughout the cluster so that in added cost and complexity and usually required larger or bigger instances for that cluster which equates to cost in addition scaling up was limited so if we wanted to scale up if we had more data we had a hard time scaling up we could not scale up beyond the number of partitions that we had as i mentioned and reality actually hit us quite a few times especially with processing bottlenecks let's see an example so here's an example of an outage that we had basically we had a downtime that started a downstrea upstream from the cluster so the producers of the data had a downtime they stopped sending data and data was piling up there this took some time to fix but when it was restored suddenly we had a torrent of data coming up and the amount of data as you can see on the graph in kafka shot up the recovery process basically started here as new data started coming in our cluster is starting processing this data and storing into the data lake and it concluded right here during that time if some of that data's retention timed out kafka would automatically delete this data so this really means if you have an outage that's long enough you lose data and here's the amount of data that that we lost in some of the outages we had so basically this is a case of that bomb exploding and data getting lost not a great situation to be in moreover the recovery time was very slow so in this example we had four hours of an outage and during that time data kept accumulating and then just recovering all this and going back to normal took seven hours so as you can see recovering all the data as new data keeps coming in takes a long time or at least took a long time with our old system why is the recovery so slow let's take a look so this is a kafka cluster example at normal times we have some data in the partitions and the consumers are reading them happily after an outage we suddenly have a lot of data but still the consumers are limited we can't add more consumers than the partition so we're pretty much stuck with the i o that we have across all the consumers and that usually means a slow recovery remember that after an outage you might need to process data maybe 10 times as fast and usually these consumers can't grow by much so we really want to scale the pipeline uh and but how do we do this so common sense solutions would be to add consumer machines but that requires adding partitions into kafka that's an option another option is to move to larger machines but again that there's a limit to that we can't keep moving to larger and larger machines there's no availability endless availability of larger and larger machines so there's really no choice but to add more partitions let's go back to the partition calculations and see why that is not such a great solution the normal throughput as we've seen before is the consumer's time consumer throughput and as we've seen the max throughput is limited to the number of partitions to consumer throughput so it seems like if we add partitions that's great we'll just add increase the max throughput again that's great but that creates all kinds of problems first there's implications for the kafka cluster performance because now kafka will manage tens of thousands of open files and that requires tuning of the kafka clusters and all kinds of difficulties around that in addition if we have a lot of partitions during those load times we'll end up with small files and we have to deal with them just as i explained before and once we go with a lot of partitions it's really hard to scale down back to a less partitions if we don't need them anymore in addition if we decide to stay with the load throughput that cost us as well in retention time let's see why so assume we need two days of data retention but our processing is really slow like in the example i showed so if we if reprocessing two days takes three days that means we really need five days of data retention in the kafka cluster so that's a cluster that's two and a half times larger than if we just had two days in it if you recall our kafka cluster which costs 540 000 a year adding two and a half times uh to its size would cost us around eight hundred thousand dollars a year just for this a retention so this is really really expensive what this really means is that if you can re-architect your system for faster burst processing you can save money by reducing the size of your kafka cluster so this is really great and there's lots of opportunity if you can do this all right so going back to our previous kafka a consumers which were using spark streaming let's look at some of the downsides to that architecture first you've seen this slide that our data is moody and it goes up over the day up and down so during the low times our spark cluster was idling and wasting our money that wasn't great during burst the capacity was limit limited and as a result the cluster was underpowered and we had those long queues as you saw in that burst example i showed you again that's not great and when we went to calculate the efficiency so the cluster as i said was not really uh working fully during those low hours the cluster efficiency was only 30 so from that respect that was not great either we decided to go out and re-architect our system so we set out with a few goals one we wanted our a the whole pipeline to auto scale and when i say the whole pipeline i mean kafka and the consumers this is the whole pipeline we wanted this to auto scale we wanted to be able to do quick burst processing for all these outages that happen once in a while and you saw why and also we wanted this to be cheaper as a whole but then there was a question how do we actually do this if we go with the common sense option of just increasing the partitions that could have worked we could have run less consumers in the in the uh in normal times and then when uh there's a lot there's a burst we'll just increase the number of partitions but as you've seen this is not a great option because of all of these reasons i mentioned before so what we really were looking for is we wanted to keep the number of partitions as it is break the relationship between consumers and be able to set our spark cluster free and scale up with more consumers when we need it we just wanted to add more and more and keep the number of partitions fixed to the optimize optimal size of our data so to do that i'm going to call ido to tell you all about the concept that we've used and he will tell you all about it right now ido thank you offer now let's go over for the juicy stuff okay so the main idea that came to us was to spread the work across isolated workers by doing that we can break the relationship between kafka partitions and the number of consumers we use so if we take a look at our stream of data we can break it into small chunks of discrete time slots we chose ours but you can use any time slot that works for you each each one of those time slots is a task and we can leverage kafka api to consume the data in kafka between those hours so we can have a task processing the data between three to four four to five five to six and so forth in our case we use spark so we launched we were using a spark clusters that each each one of those power cluster was consuming the data for uh between the time slots and when when it ends it terminates now let's see how it looks if we take a look at the at the hours the hours along the day versus the time it takes to process so we are processing and processing an hour of data right but you can easily see that some of the tasks are shorter shorter than an hour because there's no much data to process during that time and some of the tasks are longer than an hour it's all um it's all about the amount of data that got into kafka during that specific hour but even if a task takes longer than an hour that is no longer a bottleneck right because the next task will start when it was scheduled and it won't need to wait until the previous task ends what else also when the task ends we terminate the cluster and we free the resources so we are now not paying for this idle time if you think about it this is very very similar to how serverless system works in terms of cluster efficiency since we're using spark there is a small warm-up time that we are not processing any data but when we start to process the data we are pretty much around 90 of loads so the total efficiency is around 75 percent for the system of course if we can shorten the time of the warm-up time and we can get more and we can get closer and closer to the 100 efficiency now let's compare it to the previous system you know this slide right you've seen it already so in this slide we are we have all in this previous system we have all those wasted processing that we didn't need right and the efficiency was about 30 now with the new system when we are paying only for the resources we need and their efficiency is around 70 75 the new system is now 60 cheaper than the previous one now let's look at the mechanics and see how it actually work in practice so we're using airflow to managing our pipeline which means that every hour airflow spin-up is far cluster with the details of what it needs to be to do then spark go to kafka and consume the data between those hours he runs some transformation and write the result to s3 when it ends it terminates and free the resources we also leveraging emr auto scanning mechanism to terminate idle instances in case of data screw okay but we started with with burst and we started with outages what happened if there is an outage and we need to re to run for several hours that's not a problem right because airflow will just spin up more and more isolated pipelines and they will work independently and process the whole the whole time period that needs to to get processed okay this is our a airflow dock this is how it looks we basically use one pipeline that uses configuration and in the configuration we can tell airflow which topics to consume what are the time periods and any other details we need okay let's see the results of this new system so here you can see this is the transition point you can see the left side uh the where we were in the streaming and the streaming and how we went to batch so with the streaming with the left left side when we were running a micro batch every six minutes and in the right side we were running an hourly batch sometimes we might miss the batch because there was a job failing or clustered cluster crashes but that's okay because the system know how to automatically recover and heal itself and after a while it will rerun the missing task and process the missing data in terms of cost uh with the old system the cost was pretty steady along the time but when we moved to the new system now the costs change and it varies day by day because the data changed right and we got a 60 drop in cost okay this is a very cool slide because what you see here is how it's you see the data in bytes by the hour uh that the system process and here you can see the uh how much it cost uh the system how much it cost us to process that out that the data what's nice to see how it pretty much correlates and this is exactly what we were looking for we were looking for paying only for the resources we need and nothing more in this topic alone we started with 400 000 a year and we are now down to 160 a year and we actually processing more data we started with the the presentation with outage entering right and you also already saw this slide so in this slide as we said it took us seven hours to process the outage and the the to recover from the outage with the new system this is a very similar outage with it took us only 40 minutes and the reason for that is because the system automatically launched four concurrent isolated clusters and that would help us a and process it so quickly so in total we are now 85 percent faster in terms of recovery what we also gain here is that we we are now able to reduce the retention policy of kafka from three days to two days and by that uh save third of the cost in terms of storage here's another another a outage we have you can see that we were out for 20 out 28 hours and when we got to the point of recovery when we started the recovery we had roughly around six and a half billion events waiting in line now here you can see with it took us less than two hours less than two hours to recover from this outage that's amazing [Music] in terms of cluster utilization uh you can see that after we finish we with the initialization stage we're pretty much close to the 90 percent with cpu and with the server load distribution you can also see that we are very very utilized the cluster is fully utilized okay so let's summarize what we talked about we talked about the implications of not scaling up and down and we talked about the options and the challenges of scaling your kafka pipeline we've seen an example of of the of the project we've done and how we optimize it and optimize the solution for doing it and we derive some architecture insights so the conclusions are that you really want to design and build your system from the beginning uh and prepare for the worst you really need to consider the edge cases right um you really want to split the split the work into isolated tasks uh in terms of stream versus batch if you most of people won't want to go to streaming because they want the data to flow but you really need to consider batch it has its own advantages if you not really need real-time data and you you should aim to remove the dependencies and as we did with the politicians and the consumers okay so i thank you very much and i'll conclude with that you can reach us here thank you all right uh thank you very much for the interesting talk always interesting to always nice to see hardware being utilized more efficiently we have a question from a discord from uh von schaff regarding the scaling pipeline with a new server as plus spark patch approach how would you scale your kafka cluster if kafka topics have the same number of partitions the throughput will still be limited by the number of partitions unless you change the number of positions directly um so in our case not really limited kafka has a much higher io than most of the consumers because kafka is highly optimized with binary formats and it doesn't do much processing it just reads the data saves it to a uh to disk uh and then and sends it to the consumers when they read uh so as a result kafka usually can process maybe 10 or 15 times more than the consumers most of the consumers you may use uh we got around the partitions by using the timestamps because you can use the kafka api to ask for timestamps instead of asking for the offsets and then having to read from one offset to the next we actually asked the kafka api give us all the offsets in each partition from let's say six o'clock to seven o'clock right so we get the exact all the messages we need to read and then we start reading them now as we are doing this we could start up additional clusters that would at the same time do the same for the for other hours like seven to eight eight to nine and so forth so as you saw in the example uh we can have four or five clusters running in parallel as consumers and reading from the same kafka topic and partitions and i think at our worst case scenario we had how many like 12 12 15 12 it was i think 12 i think we had 12 clusters at the worst case scenario running in parallel and kafka took it uh you know without a hiccup not an issue at all so that taught us that kafka is first is amazing you can really process huge amount of data through it but also the the bottleneck a lot of times are the consumers make sense so the number of partitions stays the same throughout the yeah just the realization of consum consumption is increasing okay hopefully that answers the question or maybe there will be a follow-up but i remind the audience that the edo and offer will be available in spatial chat right after the talk let's see all right um thank you very much again and let's go offline thank you thank you we are all fine all right guys thanks that was awesome thank you very much thank you see you on discord i guess you know how many people there were attended do you have any numbers uh so i see sorry uh yeah i see uh 192 people online nice but it's not clear which which track i thought how many tracks there are two tracks right tracks yes well they are online in discord so they might as well just have discord open in the background but that's uh also some of them might be not in discord so maybe uh maybe that means we'll know uh all right super so we're going over to discord i guess we'll do uh no you're supposed to go to spatial chat special chat sorry uh do we should we disconnect from the zoom yeah please describe chromosome all right super thank you thank you