Scale by the Bay 2018: Gwen Shapira, Matthias Sax: Deploying Kafka Streams Applications...
you hello welcome to our talk yeah what are we going to do today quick quick agenda first of all I would like to give an introduction to a Pesci Kafka any particular Kafka streams so I guess who is using a Pesci Kafka and know Patrick after almost everybody who knows Kafka streams of people not bad yeah so for everybody else we're going to give a high-level overview so you're not lost in this talk what's going on here and then I will do kind of a deep dive and explain how Kafka streams applications scale in particulars this is about state handling and that's important to understand because in the second part when discovering the kubernetes using kubernetes to deploy those applications and using combined idiots you have certain options how you can deploy an application and state is a key part that's important to understand how Kafka streams handle state in order to be able to make the right decisions using communities deploying an application like that and in the end when is giving a live demo so we hope that works works well and yeah let's get started so as you know Kafka has three main components there are sea brokers so the Kafka cluster say this is a storage layer in the system and there is Kafka connect that is used to import an export data from into other systems and sense there is Kafka streams and Kafka streams is Java library said you used to build stream processing applications and even if it's written in Java we also have for half a SCADA API for that said was contributed in SE tu lado release bye-bye light band so we are very happy to now support not just Java but also Scala in a native way and while you write this application and sense the question is of course well if you have it you need to deploy it because it's a regular public it's nothing special about it and in this talk we are explaining you how you can use kubernetes to do this and so this application is connecting because the copper cluster it is reading data streams it's processing the data so you can filter you can aggregate you can join multiple streams and then it's writing the results back into the Khafre cluster so let me give an example so you get a look and feel how you can how the APR looks like and this example is example said when is using in her demo later on and it's going to be about doing statistics over stock trades so when when we start this application and I left out a couple of details about implicit sand all those kind of things just like says a coursing you have this stream spill der sets you mine you could say like a factory and where you basically express your application logic and that's all like lazy so because it's a builder so basically you first express your application steps and then when you're done you basically kick off the computation explicitly so when you have this builder but first you want to read some data from the Khafre cluster so you say here well please create a data stream that is a case stream that's a return type and reads data from this topic stocks and when we when we have this data stream now we can start to process the data set it's flowing in the stream and in this example we would like to compute statistics so we are having to do an aggregation of the upset and aggregations in Kafka streams are always based on keys and so that's basically value model we inherit from Kafka and doing aggregations based and he's also allows us to scale out the aggregation obviously because it's sent data parallel so here we say well we want to Harvey one a group the current stream is a given key and we also apply a time window because while a stream is infinite and doing an infinite aggregation is usually not desired so you wanna junk the stream into finite pieces and here we get a window of a size of five thousand milliseconds or five seconds and we removes the window in one second intervals and then when we have all those windows while we apply an aggregation function that means for every new window we are getting we are creating a new trade statistics object that is holding our statistics if we want to compute and then for every record in the data stream or in the window because the aggregator function that says well we are getting a new trade from the data stream and the key here is not important so we ignore it we have the current aggregation result that is stored in the trade stats and then we just said well add trade to the statistics and what's the trades that does internally we don't really care in this example it's just like adding it to update the statistics and the important thing here is that this aggregation step is now returning something different it's not a stream anymore it's a table expressed in this type of K table and saying kafka streams we have this notion or is this concept of stream table duality mining a stream and a table are the two main abstractions we're providing and a table is now introduces state so we're entering the world of stateful stream processing and of stageful applications and that is why also state handling I'm going to explain later on it's so important because well a table is backed up by a local state so then we can also transform the state actually back into a changelog stream so for every update you're getting via getting an update in the output stream we map the values again to actually do our computation so here we want to have the average price and then we write the result back into an output topic and note up to this point it is all just like the clarity it's just saying what we're going to do no computation happened so far and now in order to kick off the computation via creating this Kafka streams object that you can think of as a client and you're giving instability some properties to configure so for example how you can find the catalyst and things like that and then we call explicitly stream stock start and this will kick off computation and now everything is running so an important part is here's this build about building because as I said everything is lazy and you express this as a questions how is this all act act actually executed in practice policy is using this API for project by showing from black belt who's sitting somewhere or thank you okay cool so yeah so we call it the spirit of both and now we want to want to look into the engine room how this is actually executed so this program consists of three main parts so we have the Builder of stream then we have group by window aggregate map values and - and internally everything is represented as a dag of operators that what we call a processor topology and well for all of those three mines for C steps we creating one one processor node so we have a source node we have some processing nodes we have a sync node and well data is flowing from from node to node so we have data streams connecting cells nodes and and as I mentioned this aggregation is giving us a table and so for this processor node we need state said we need to handle so we also create a state store so that's a basic thing but how do we actually now scale an application like that and and for this we have to we have to look into how Kafka works and Kafka has this notion of topics that are partitioned and a petition is a unit of parallelism Kafka and that is what Kafka streams inherits and in particular we are representing this in so-called tasks so let's assume we have an input topic for petitions and so we can create a task and the test is running a copy of this process of topologies that I just described earlier and because in this example we have in for input topic partitions well we're going to create four tasks and each of these tasks is connecting to exactly one partition and consuming the data from it so it's fully data parallel and what's also important is you might be familiar with consumer groups a Kafka streams application it is actually consumer group uses a consumer and that is how we can can coordinate the assignment from the petitions to sales tasks and so that allows us how to scale out and so in this case we could have up to four threads running in this consumer group so you can have either four instances and four different machines or you have four lowest rates in one machine each processing one task and well so of course the result is also written back in two thousand and and it's important here we basically have this and source we have state before I forget this and and state is now charted so for each task we get one charge of the overall state that is representing the sky tables that I described before and it's important to understand that series this is one to one to one mapping so we have one input topic partition we have one task for it and we have one shard of the state and that's basically atomic parallel unit of processing in Kafka strings so now with this how do we scale the application so let's assume you're starting a new application initially says of course no state means the state is empty and because we have four input topic petitions well we would have four tasks and also for tasks are assigned to this instance and during processing well now all this instance is building up state and we see I color codes this year now is different colors but it's the same state it's just like the different charts now when you say I want to scale out my application you can say well I start a new instance and based on the consumer group rebalance protocols that were using underneath cells instances get aware of each other we would reassigns input partitions and now we would also move the tasks so it means in this case we could say we move two tasks into the new instance and in order to to preserve correctness we also need to move the stage from instance one to instance two before we consume a resume processing so those two instances also gets a corresponding shards assigned and notice that it's important the state on instance one is not immediately deleted we keep it around and I will explain why we do this later on but that's important to understand so basically we get we get copies of soft shards in instance two and now we can even start one more application instance we do the exact same thing and we saying in this example we move this as a green chart we rebuild it and now we're in this state okay since it's great another important thing is is for tolerance and so what is happening if something goes wrong because it's distributed application and what's the nice thing is scaling and for tolerance is basically the same thing because let's assume we are India in this state here it's a little bit different I remove this as a yellow shard from instance one and let's say application instance two fails based on say consumer repellents protocol sis failure would be detected and we would trigger a so-called rebalance to reassign the green as a blue and the yellow partition two different instances and now we have we have two choices for both we can either assign them to instance one or two instance three but now as you can see instance one already has a part of the blue state locally stored and that is something we can exploit so Kafka's dreams is being a little more than plain consumer it encodes this information into the Rebellion's protocol and this helps us to makes a decision to say well we want to move say says there's a task that is responsible for for the blue state into instance one so we don't need to recover the state completely because we have it already partially and we would move send for load-balancing reasons says also states a yellow state to instant three we would first have an empty states if we first need to recover and then we can start computing okay so I talked a lot about like state if moved from instance one to instance to to instance three but how does this actually work because as I mentioned it is lower state and while there's some magic going on here and how we do this I want to explain here we have a give a single instance of a task we're reading an input topic it's processed we have the state and series our topic so and in order to make this state fully fault-tolerant we create a change of topic in casper cluster and every state update that is done is also written as an update record into this change of topic and that allows us to store the state reliably so basically means the change of topic in the Kafka cluster is the source of truth and the lowest state can be seen as an ephemeral cache so even if he writes this to disk writing to disk is not part of the for tolerant story the four torrents happened in the Kafka cluster because there's the topic is partitioned and also replicated so you'll never lose the state and when we need to recreate state well we can go to this change of topic and read all those updates in order to recreate the table so how does it work in in practice will be migrated state here we assume we have we have a single instance with all those four tasks and as I mentioned before we have this idea of of atomic unique units of parallelism and that is also reflected in the changelog topic because we have four charts or four tasks we are also creating a changelog topic with exactly four partitions and we do again this one-to-one mapping of this chart is now extended into those change topics so for every shard we use exactly one partition we don't do like hash based partitioning or whatever we just assign it one-to-one and do it manually so why is this application instances running and it's updating it's a local state it's also write and also change it into the change life topic now when we scale out and we create a new instance and those tasks are real science the first thing this new instance is doing well it's going to change the topic it's reading the changes and recreates the state and after that is done it goes into processing mode and now it writes into the change of topic so let's talk about but recovery so one important thing you have to have to know about the change love topic is said it is a compacted topic most people are familiar in kaif curve is the concept of retention time it means well you store data there and after a certain amount of time the Khafre cluster is deleting it if you would do this for a change log topics they would be dangerous because let's say you have an update for a certain customer said you put into your table and then this customer girls doesn't use the application for a certain amount of time and then it would be dropped from the change of topic you would lose this state it's something you don't want to have but caf-co not also health topics that can be configured with retention time but also with log compaction and for log compactions there is no concept of retention time so all data is stored forever however for for every key value pair the only preserve the latest the latest entry basically if you have one customer and you write like three updates into it into the change log topic since the cluster is allowed to garbage collect say all the tools and keep only the latest one but the latest one is never be deleted so even if a customer goes offline for a long time or doesn't use replication you will never lose the state because it's never never updated and this mechanism when this topic is compacted and guarantees also said the size of the topic is linear in the size of actual state because if compact is perfect well you will have exactly one key value pair Turkey and now no old updates it also means of course if you have large state if you need to reread this whole change of topic you might have high recovery times and that's also the reason why we say we want to preserve state and the application for longer locally so we don't need to go to the change of topic and reread it so this state that I described it's not just in memory also written to local disk when if your application is offline and comes back well it doesn't need to go to the change of topic because it can just access local disk and say I have the state already so for large states that's usually not a problem but if you have small states there is one one Watchers that I also want to explain here so because topics our further divided into segments or actually topic partitions and the default segment size is one gigabyte so if you have a large state let us unite 20 gigabytes and you right into the topic since there is a active segment there is a new segment where you append new records any of all the other segments and as long as no compaction is not not triggering in so it's basically a background process on the copper cluster well you keep appending and appending and appending and also you change the top you can of course grow a little larger than your 20 gigabytes and but when log compaction takes in and all the segments are removed or basically reduced in size sense and if you configure low compaction aggressively enough you know in average your your topic size will be about 20 gigabytes plus this active segment because the active segment is not considered for low compaction only the old segments are compacted that means if you if you have updates in the active segment you can still have duplicates in this interchange our topic until the segment is full and we create a new active segment and census garbage collection process removing old updates would have so for large states it's not uh not a big issue but if you have small state that can actually impact your recovery time so let's say your state is only 100 megabytes here and you still run the city fault segment size of one gigabyte after lot compaction kicks in it can happen that your overall topic size is still one that one gigabyte because the active segment with one gigabyte is not compacted and the rest is of course compacted to 100 megabyte but it means basically is that you said you said you can have like up to 11 duplicates of your key and your recovery time is much larger and we actually have seen this in a customer's it had a smaller state and servos kind of allows data so small and recovery takes so long what's going on and that was the exact root cause and we told them to decrease the segment size and said speeded up the recovery time from like 10 minutes to 30 seconds or something like that so for small states that's it's important how can you use this or in general so the overhead is proportional as a rule of thumb to segment size over the state size means if the segment size is smaller than the state size then you get reduced overhead so that is what you want to have and there are two configurations for that that you can use the size of source log segments and then also how aggressively you trigger the compaction and that's also configurable one last thing I want to mention is though you should not be too aggressive itself settings so you don't wanna make the segment size too small because smaller segments is more overhead on the broker side so say it's the sweet spot between like overhead and the broker and recovery time said you need to figure out for your own application okay and now I'm handing over to Brent and cheese explaining you how you can use all this within Cooney's so basically we now we know how it scales in theory but at the end of the day you are deploying stuff to production and how do I actually do more instances and make this thing scale and go up and down in production is a totally different question and many of us these days ryan kubernetes productions so I'm going to tell you how to do it with kubernetes and in order for this to make any sense um how many of you use kubernetes and totally know what I'm going to say okay I'm not going to bore all of you on this some of you kubernetes itself is super easy so nothing to worry about it's a database it's a REST API and it's a scheduler so you talk to the REST API and you say this is my configuration it stored in another base and the scheduler in implements your configuration that's it you're now all kubernetes experts almost I'm going to dive into mini this is the thing that makes communities difficult other than the usual storage and network is the huge amount of terminology around it's like everything has a lot of names to learn so I'm going to walk you through few names that you need to learn in general you're the basic unit that you worry about is your application now your application has some dependency and it has some configuration and it has some environment variables and it has some CPUs is that it's gonna need so most of us container as their application and we think of the container is the thing we are going to deploy not so kubernetes which does not think of your application or your container is a thing it wants to deploy kubernetes wants to deploy pods a pod is a set of one or more containers that are always deployed as a single unit it has storage which he may be persistent or ephemeral and it has an IP which is always ephemeral meaning that you kill the pod a new pod will start it will have a brand new IP it will have brand new storage and if it may run the same application but for all intents and purposes it's a totally new thing now one pod is our application but usually if we think about things at scale we think about collections of pods because we need our web server to run 50 times or so and in order to talk about collections of posters the old terminology which is still very much around which is replica set a replica set is basically a way to identify specific ports for example all the pods that are running the stock start application in production I 5000 of those versus all the pods that run the other application in dev and they only need two of those this kind of thing this is a replica set that's all terminology new terminology is a deployment which is the same replica set but it also comes with a policy for example you are only when you restart them set them one by one versus start all of them together is that kind of thing now when I just remember when talk about replicas I'm talking about kubernetes replicas as opposed to decouple replicas we have a lot of replicas all over the place now we talked about stopping and starting those replicas one of the things that you probably care more than you think you do about how those replicas are placed on physical machines for example suppose that I take my calf customer's application then I say I want three copies and by chance or maliciousness all three are placed on the same physical machine my high availability plan is not as good as I may have wanted it to be right I have certain single points of failure zin my plan if if I could tell you bananas hey the pods with this name never ever place all the replicas on the same machine actually just one per physical machine please life would be a lot better for me and I could actually have a feasible failover plan in place this is the affinity and anti affinity rule and the affinity is bless them far far away from each other affinity means these things should actually run together for example if you have computer storage or that kind of thing you may want them together now there is also question I mentioned that IPS are ephemeral which is kind of sometimes okay but most of the time our services need to talk to each other so I have a front-end that it has to access the back end but if the front end if the back end only have a firmer IP how can I ever find the back end server to talk to who do who do I connect with so there is the concept of services so basically create a service name and they do tell us the load balancing and outing behind the scenes so my front end says I want to talk to Ed back-end I have five pots from this back-end running around this replica set and it knows about the ephemerality and routes my request so it's kind of a DNS and internal router and load balancer kind of wrapped into the service concept now this is nice for sinks when you don't care who you're talking to but in case of something like Kafka you probably care deeply about which half co-broker you're talking to because Kafka has States this broker is different from the other broker this one has a leader for those partitions and if I want to write to this partition I need to talk to this specific broker know as a broker is going to do and when you have this kind of identity where the specific application service that you're talking to also holds the storage that you're interested in and it has a specific access point this is where you get to the concept of a stateful set so there is a lot of different ways to understand stands for set and the state itself is just part of it like the fact that it has storage and has a persistent volume is nice but it's not the whole story the whole story is at each one of those pods there are no longer random pods that these are port one which is different than port 2 which is different than port 3 each one of them has their own unique IPS they can talk to each one of them has its own unique storage that is persistent for the one always connects to the storage of pod one always has the same IPS pod one they have their identity this is stateful sets which is kind of the core concept of having to manage state inside kubernetes ok with all this knowledge about how to do kubernetes what do we actually do with Kafka streams so as you remember we have those applications and they each have state so one may say that ok is a half state we may need a stateful set and this would be a reasonable thing to say hey here is the state in the photo let's use a stateful set to store it which is kind of correct but it's also correct that if we it's an ephemeral state right if we lose the state we can recover it we don't need to take tons of care and do a lot of extra effort to make sure that we never lose it because we already know the actual state the real the durable one is in Kafka it's basically a cash who cares about caches like and a lot of things have caches and which throws them away all the time so you can reasonably say that heist at full size are kind of a pain in the ass and we don't need them and why even bother but of course as we learned what if it's a large state well it takes time to recover do we really need to spend all this extra time recovering the state where we can put some extra effort into having state full set and get to keep it around this is Humber's and true but of course if you're going to scale in and out a lot then you're not going to really get advantage of the fact that you have state you're going to have to rebuild it every time scale out and then you lose it every time scale in and also if you're going to have stateful set well local storage for stateful sets is still in beta which means that you'll probably want to use shared storage if you're in the cloud if you're in AWS GCP Asscher etc it's probably fine if it's in your own data center it's kind of a contour some of us has fantastic storage means and fantastic storage and everything is fine and rest of us may have a storage it means a certain antivirus or make Africa storage at 3 a.m. in the morning every Wednesday and I have to wake up and rescue my cough a cluster this may or may not have happened in real life so sometimes you want to stay far far away from the shared storage for political reasons so overall recommendations no matter what you do take materials recommendations to heart and try to keep the changelog shard small it may mean having more partitions in Kafka in order to have smaller state if you have good shared storage to contrast state there's no low real drawback to using stateful sets and show you it's not that difficult just do it they use anti affinity that's important you don't want to put all your eggs in one physical basket it will not bow dwell for your recovery strategy the other thing you want to use is what we call parallel port management which means that if you are going to start your stateful set do we start one pod after another or if you do we start them all at once if we start one it will create a state and then we start another one it will have to create a state again and then we sell another one it will have to create a state again not a good idea make sure you start all of them in parallel shut them down in parallel so that's important and then there is the new concept I learned about it relatively recently of disruption budget which is basically under normal maintenance how much how many of those a No pods are we allowed to take down at the same time set to something reasonable like one one something a good number just make sure that if you do if you upgrade your kubernetes it will shut them down one one by one and not all of them together unless you actually schedule downtime in which case everything is different okay so this is all cool but how does it actually work so I'm going to just walk it through kubernetes for those of you who had the pleasure of not going it's configured with very very large German files so prepared for a lot of Fierman and let's start with the deployment which is stateless and we basically say ok we're doing a deployment and we are going to call it stream stock starts because that's the name of what I'm deploying and I'm going to start with a single replica I can scale it out later if I wanted show you how to scale it out and basically I'm selecting pods that have the running application called stream stocks that I'm creating the pod right here from this template and I'm saying it's going to have the application called stream stocks that's I'm assigning their anti affinities so they're not going to run on the same physical host or I'm running on gke so it's not a very physical physical host but somewhat physical physical world and then I'm basically saying please deploy this docker container which is available on my repo that's pretty much it the anti affinity part is actually pretty bit much longer in reality but that's the main idea and as you can see it's not super complicated and it's available on github now if I want to do a stateful set I started by declaring a service and in this case I don't need to add a piece because I'm not going to use them and then I'm saying I have a stateful set here let's run two replicas start them in parallel please and I'm saying I want volumes because this time I want persistent storage so please create volumes for me and please mount them in this password Kafka streams expects to see it state a give them a name and I want to have one gigabyte for each one of those they say storage volumes and we'd write one Smisek one pot is allowed to access this volume and read and write to it I don't want it's not a shared volume with specific for each pod so read/write ones right here and that's about it very straightforward input anti affinity but you probably should put on the affinity I didn't put the disruption budget because I just found out about it and frankly I don't know how to use it yet but that's more or less the idea and it's fairly straightforward so you have whoever tells you that stateful sets are super complicated it's probably because it's storage Ivan M yeah we have confluent opera I'm doing the my demo on the confluent a cloud but if you're running it yourself and running the brokers you can use confluence operator for that now in better and let's see if I can get a demo going do I have time for them no you're not holding it this time come on no it would be silly well can you still hear me can you still hear me quarter can you say something that looks like terminal yes I think you can let's start looking at what kubernetes is running which is right now nothing so let's start running it's going to be up slightly scripted M I'm going to start by running the producer basically just going to produce the trades on the to the kubernetes cluster to the castle after so I'll have some statistics to a process so here we are starting the producer let's make sure that we're actually producing something ok so you can see season is basically I'm looking into the topic and this is what the trades look like and you can see that they're coming in so I'm getting new trades they have a ticker they have a price they have sizes they have all kinds of stuff in them this is nice next thing I basically I want to create my deployment and I'm creating a stateful deployment running the ml that you've all just seen and now is when I want to see them so you can see that I have two pots and they're kind of in the pending state is they're about to start and you can see that one of them is called - zero and one of them is - one this is how you know this is a stateful set if this was a deployment it would be - longer random hexadecimal a string and - something else Randall and if you'd kill one of them you'll get a new random number because it's a totally new thing with new IP and new storage etc well here if I kill Stokes at one I will get stocks at one back with the same external IP in the same storage etcetera this time no IP because it didn't configure one but you get a drift so at least one of them started the second started let's see if we have results so let's go to the other topic stock stats output are you crossing fingers for me command please be processing something M okay I guess it isn't one of the things I can look at oh here we are fantastic so you can see that I have the sum the minimum driver price we have statistics guys and I can also go and look at the application kind of look at the lag like how far behind I am and this is a good way to see as I'm rebalancing and creating new way scaling in scaling out is it still processing am i still keeping up so let's go back and try to scale it out so first of all here you can see my volumes and now I'm saying please scale and I now I want three replicas instead of two so if I go over here you can see that now I have another one of those starting up and if you can if I go back here you can see that it basically keeps going and processing I still have only two it didn't so I think one of them is still on the way to starting up but didn't completely start the idea it's still creating once it gets started I'll be able to see it scale out so now it's running so let's see if I actually have three of those yes so now I have three of them you can see that each one of them have a different ID and you can see that it's shrank all the way down it had to stop rebalance and start again because it had to recreate the state now if I go and kill one of them it will actually start from an existing state so I'm deleting one of the ports it's basically identical to crushing it and it will automatically you can see it's basically terminated and now it's starting again you can see that in this case it's basically key we keep going because we do not we don't have to rebalance we just or in this case we may have had troubles but it will immediately pick up the existing state and keep going so that's about it I mean the next thing is just to take down the entire say bye-bye to my demo and call it a day so yeah thank you everyone I mean I think we're out of time so if we'll be around for questions I think until the next talk and even afterwards so thank you you