BDSBTB 2015: Vlad Giverts, Scalable Search with Kafka, Spark Streaming, and Elastic Search
alright thanks Lexi so I'm glad gibbets and Lexi's given most of my introduction I'll talk about a little bit more background in a second this talk is going to be somewhat high level but I'll also talk about actually quite high level but I'll also talk about some of the things that worked and didn't work for us and maybe save some room at the end for QA you guys want to have another level of detail that I don't get to in the talk all right so let's get to it so I run various data teams at work day data science data engineering you know in search as well in fact seems like in a past life now used to be the search architect at tagged it's quite a few years ago when tagged for those who don't never heard of it used to be the third largest social network in the United States number one was myspace number to his Facebook and then it was tagged so that kind of dates it a little bit and in fact back then we probably had more users / engineer than facebook or myspace ever had it was actually quite a lot of challenger responsibility in a very small team and I was the guy task was getting searched the scale so i ended up building you know solar before there was soul or at least before silver cloud and elastic search made a large scale is served a search possible that basically a one-man project so I know fair bit about search and how to make it available and scalable from those days I was also the CTO had identified more recently and identified was aggregating billions of social profiles from all over the web reconciling them trying to create this single identity of record for people and then we were allowing recruiters to search for and contact people they wanted to hire and we had to get very very good at the search doing search there as well because we were competing with LinkedIn so the bar is pretty high and now finally I'm at work day at work day and largely working with bringing kind of large-scale machine learning into production you know it product heisting answering business questions but one of the other things i'm also overseeing is search so for those who aren't aware workday is a cloud software company similar to salesforce only work day does HR and financial software and for people who use workday search is more often than not the primary interface the entire application so if search isn't working well then they're not going to find what they able to need to do and they won't actually be able to get work done sexually mission-critical and we've done quite a lot of work to Rio protect search within the latest and greatest technologies to make it work well that's kind of my background on search so today we're going to talk about search scalability and also availability when we talk about scalability what do we mean obviously this is a fairly sophisticated crowd so I'll go over this pretty darn quickly it's not about how much weight you can put on something though and that metaphor could work as well it's typically about how high you can go and in this case it's more about scaling a mountain that's typically and when you see those charts you know that go up into the up and to the right that's usually what it means by scalability whatever metrics that you have allow them to continue going up into the right whether it's you know computation or memory or storage or networking how do you get that to scale and let's use if you know really simple metaphor office buildings you can only cram so many people into a building in order to get the work done and if you want to get more work done in the same kind of like two dimensional footprint that building is in what he gonna do well you're going to scale up because that's yeah what people do that's you can see this going on right now in downtown San Francisco and that can go pretty far so it's called vertical scaling and you know in the case of computing you know typically you know try to get whatever it is that you do to operate within the same physical infrastructure well same logical infrastructure a single node single server so you go vertical and you make great this you have great big monolithic nodes that servers what have you that can then have more memory more cpus more disks whatever maybe maybe you attach one giant disk array if you need more I ops that's historically bid how scaling has worked unfortunately oftentimes scaling looks more like this well yeah I'm not not sure how much more vertical scaling you're gonna have room for and at some point the applications will reach this point very often sooner than you realize and it's often very expensive even to scale up because the hardware i will get or complex more exotic so you know there's got to be a better way there's horizontal right now go you've been in the south bay you've probably seen this at horizontal scaling the whole suburban office parks just spread it out so it's much much cheaper to spread out than it is up if you look at it probably building two of these buildings probably a lot cheaper than building one that's the staff and top of each other particularly when that one is in downtown San Francisco so you just keep going you on triple capacity no problem get three nodes quadruple capacity for notes wonderful no problem well actually the problem is this software is that incredibly difficult to write and and we'll talk about a few piece of software they do this well but I've done quite a bit of distributed computing in my time particularly a tagged and getting distributed right is it's no small feat what about availability what are we talking about what I mean that well got one of our workers trying to go to the office building and oh it's closed closed for maintenance maybe is a termite infestation there's some kind of bugs bugs in the building what do you do Wow if you had in building he could go to then the worker could always get his work done and this you know same metaphor you go there instead in fact have three buildings now what are the odds at two out of the three or all three are going to be down at any given time or down for maintenance extremely low and typically this is the story to availability with availability you want something truly highly available you could try to make one program that's super resilient tried you know building a building that you never have to close for maintenance never gets any bug infestations that's very very hard it basically never happens you will always go down at some point so if you can have more of one whatever it is you can guarantee that's when you can guarantee availability and guarantee up time so what are we talking about it actually today well at work day we're doing search and here's what the architecture looks like top of the stack something called OMS object management system is this homegrown in-memory data store I won't go into that oh then that's sort of the system of record that we operate with but then the next levels down the staff is what really i'm going to talk about next level down is Kafka so the way it works is whenever any piece of information that we care about indexing certain changes in this LMS it sends a message to caucusing hey here's some here's a new piece of data please update the index with it then that gets handed down one level down by two spark streaming so spark streaming is continuously reading from Kafka and pulling in that data re indexing it and pushing it down to elasticsearch and this this works reasonably well we're able to get you pretty wide scale this doesn't this really no limit there are some limits not talk to them but it does scale the way one it is highly available so actually let's let's dive into this and understand exactly how each component works so first there's Kafka so will Thomas Kafka scalability that's probably it's one of its strongest suits when requests come in to Kafka what happens well first a client has to make the request and that know the reason I'm calling that out how about that in a minute and really client can make requests to lots of Kafka's servers you could have as many as you need so if you have a huge volume of traffic you're going to send it to you know maybe to Kafka notes3 Kafka notes however many it is that you need and I realize we just had a Kafka talk so this I'm going to go through this very quickly of course you could have arbitrary many clients and they could all be talking to Kafka as well and and how does the client know which server to talk to so this is not a load-balanced service that the clients have to be smart and they have to know what Kafka server to talk to well the way they know and actually i'll get to that in a minute so the high level you have clients distributing their right load to a very large copper cluster as big as it needs to be the availability story is kind of similar you already have these several kaka nodes and now you could replicate so you write to one and then the data gets replicated to you know anymore so this what in this case we have a replication factor of two and this is really nice because this guarantees that well I lose one Kafka note how these two Kafka knowledge it doesn't matter you could actually lose it'll probably losing two of the same partitions is very low which means you're highly a pretty strong guarantees that you're never going to lose data and you'll always be able to then reread data out of that cough cook you afterwards and there you go that's the node with our data and how do we know what nodes to go to both for reading and writing and that that's where zookeeper come ins and the client first toss the zookeeper says hey you know give me give me the data about all the kafka nodes what topics are on each of them and how are they partitioned and based on that the client can then make a decision as to what cough it server to go to and Kafka syncs up with zookeeper so the numbers are the change whenever a new node comes up or a node comes down zookeeper will you know we'll see that the kafka nodes are coming up and down and propagate that data to the clients accordingly to the clients always have a near real-time view there's some lag but they have a near real-time view of what's happening in the cough McCluster and they can make adjustments that also means that a transitional state you're gonna have clients that are slightly out of sync with each other out of sync with reality it all tends to converge reasonably quickly but then and then yeah you could have you gaze with your whole a cluster every single Kafka notes talking to zookeeper and you have lots of zoo keepers so zoom keeper has its own availability story you have arbitrary many zookeeper notes and actually this is an interesting bit the way it works is all the nodes sync with each other so an update you can come into any node and it will then sync that update with every other Razoo keeper node so this means that you know data is highly available right is it but is it scalable well it's a rhetorical question but hopefully you're thinking about it it is scalable for reads 9 you do have more nodes that you can read from but because all data is replicated to all nodes all the time it means that the rights are not scalable means all rights essentially have to have them in all nodes and in fact the default configuration of zookeeper means that the rights actually have to be persistent to disk before they're considered committed I think there's a flag called force flush that you can adjust and if you don't adjust this flag you will hit your right bottle you potentially hit them pretty quickly if you have a large enough cluster with enough activity on it you can pretty easily knock zookeeper over and this is something we've done a number of occasions so we're now very careful with how we treat zookeeper but you can also tell zookeeper to not flush up in the disk on every single commit and but by the way just to understand how expensive these rights are I don't know if this has changed I haven't looked at a zoo keeper recently the last I checked zookeeper would literally saved the entire state to disk every single time so it's actually very very expensive right operations so we were backed by SSDs and we had a ton of throughput it would still be like knocks to keep her over because of too much right load but you can tell zookeeper not to persist as part of the commits and then it'll eventually save at the disk but it also means there's a significant delay generally that's okay because zookeeper is replicating to all the other notes if you lose as long as you don't lose the majority of your nodes you're not at the same time you're not going to lose to have any data loss but then you're relying on data being in memory so if there's a power outage and everything fails simultaneously when you come back up you will have lost some of your data if you've turned that forced flush flag off so that's Kafka and zoo keeper let's talk about spark streaming and really we're talking about spark because you know spark streaming that's just you know abstraction layer on top of spark and spark is a great scalability story obviously almost everyone this room probably is well aware of it you could have lots of spark nodes there's the our TVs which are resilient distributed data structures distributed across the entire cluster and they all can read from Kafka and as the data comes in they make sure that they get that processing done and in our case building search indexes or in yo building search documents and sending them to elasticsearch and in terms of availability if you lose a node no problem particularly contact with sparks streaming that data will eventually get reread from Kafka by another node and that node will then come you know build that document or perform whatever computation you wanted to perform and that will work just fine and now elasticsearch elasticsearch has a really interesting scalability story in that it's you know very elastic but you know requests can come into any node it's really convenient you just stick the entire cluster behind a load balancer and if it's the wrong note it'll just forward the request to the appropriate node so that's both for reads and writes now often wouldn't do a query it's actually going to have to query every node because if you're saying oh I'm quitting some index this index maybe is probably sharted and these shards may be distributed across every node in the entire cluster and if that's the case then you're going to go and essentially perform some subset of this query on every single node in the cluster and then whatever node you haven't initiate the request on will aggregate the results and you know return whatever the you know sort them appropriately applied appropriate search relevance scores and return the subset that you care about works really nicely to the yeah I just got built-in MapReduce too great a lot of people actually using elasticsearch as an analytics engine I think it's actually what works very well as a nest in the analytical database because you can distribute your queries so easily across very large data sets and let's look talk a little bit more about sharding on elastic search the way it works is you have primaries and you have replicas and like Kafka you could have you know as many replicas as you want so you're not limited to just one you could have you know two or ten whatever in our case we typically operate with two and again if a right comes in and says oh you know this right belongs to shard number seven then that node will then make sure forward that right to the appropriate to the appropriate node which will then I'll also talk to the replica end up doing basically same amount of work on both the primary then replica except one is the authority and coordinate the the persistence and then what happens if you lose the primary well the nice thing about elasticsearch is it will automatically promote the replica to the primary and when the right comes in and we'll go there in fact what uh let's see what I don't have a slide for it is it will also automatically rebalance so if you just lost one of your nodes it'll go and promote one of the existing nodes to you know it'll say okay now you're responsible in addition whatever you're currently responsible for you're also responsible to be the the flood recalling the replica for in a primary shard number seven and in that sense it's really really easy to administer in fact it's one of the easiest to administer distributed systems and because it doesn't doesn't have the dependency on zookeeper it doesn't have yeah you know doesn't become that well it doesn't have a scalability bottleneck the way zookeeper as a scalability ball net with the with respect to updates so that's the architecture however just a few things that we've had from it we're not we're not actually using this exact architecture anymore so OMS is still here yeah worked a system of Records not going anywhere and we're still using elasticsearch to power search but we've recently asked staff t'ride submitted this talk we actually stopped using Kafka and spark streaming and we had a number of issues for one thing we we were never able to get our queue times down below you know let's say 5 210 5 to 15 minutes and we couldn't quite get like up to the second speed see how am i doing on time we and we never quite figure out why we tried adding more memory to spark we tried adding more compute resources more workers a bunch of parameters couldn't quite figure it out there were other issues too for example I think this has been fixed now but for a while the spark stirring process would just stop accepting new work after seven days six seven days hard stop it also sometimes it would crash in the first day or two just randomly so we had you know I think it's still early i think is very promising we like the simplicity of this architecture we liked how in theory could have given us damn pretty you know good real-time performance but it never quite worked out that way for us we've since migrated to something based on a queuing system based on Redis and actually a pull versus instead of a push so we're queuing work to get done on Redis using actually a ruby gem called sidekick and and then we're just pulling over that work and quickly you know pulling the data from the LMS to be indexed and sending it to elasticsearch and I'm having a dive deeper if anyone's curious about some of our learnings with Kafka and spark as well as yeah I found in the QA and by the way we're still using spark very heavily just not for the search context where we use spark also as our primary data pipelines we do quite a lot of you know will you do a lot of machine learning on my team and we spark for you know massaging the data bill and do the feature engineering model training yeah that whole that whole end-to-end did engineering process alright thank questions yes that is correct oh so so sorry the question is to clarify that we have this stack and we've booted sparks streaming out of the stack we've actually also booted Kafka so we're we have something else in the middle it handles that that function as of right now so it's a great question so the question is where are we having the issues to be just to be absolutely clear are having it with just spark with spark and Kafka or you know some combination of the two we were not having any issues with kafka kafka was working at very well as advertised as far as we could tell the issues were with sparks streaming specifically the reason we would also remove Kafka is that in addition to this we had to have we had to build a system where we would pull data and under certain circumstances from the OMS and shove it index and shove into elasticsearch and since we already had that system we adapted it to handle the kind of the more real-time use case of getting the updates from as they're happening so and that system didn't rely on Kafka so kind of Kafka went left just by by being as a coincidence yes so what version of sparks dream we were using we're reusing you're saying the custom receiver or the the kafka API so we were built using kind of the built-in kafka facilities within sparks dreaming we weren't using the very latest 10 a new one recently it was a recently released we're using the one before before we shifted from what I've heard the kind of the hanging after seven days issue has been fixed in the latest version but I don't know about some of the other issues like the you know it explicitly long queue times that we weren't able to get that get the queue depth down below a certain point yes ok a good question so the question is are we using the last of using elastic search for on this storage and if so are we having issues with it so the answer is no we're not using it for on this storage in fact we run a hundred percent in memory the workday has these odd requirements not so odd if you understand the depths of it but every one of our customers data has to be encrypted if it ever gets to you know ghost goes at risk at rest on a disk which meant that we couldn't even use the storage capabilities of elastic search so we're running like Ram disks for elastic search which is actually a problem because so some of the issues we have had was if we tried rebooting an elastic search node while that node was restarting elasticsearch would then immediately try to rebalance and if we were trying to do somethings like you know just upgrade elasticsearch many potentially restarting the nodes actually if i specially did too many of them would forget the cluster really confused and into a bad state the recently elasticsearch released a new version change that functionality so now when you restart a node in elastic search Network start when elio disappears it doesn't try to rebalance immediately it waits a little bit which is good things now when a node comes up it can reassign the same shards that that node had before it went down that prevents you know all this you know cluster rebalancing Andrea and shuffling around that can happen when you rapidly RIA restart nodes yes so would flavor spark you were using standalone or what mmhmm yeah are we using a cluster manager yes a good good question we were using spark with yarn so we before we move to spark we were using Hadoop with scalding and that worked very well for us and we were already using yarn with a dupe so now we just you know what we know how to operate Hadoop and yarn let's just run that one spark on top of it that's worked really well for us it so the questions are the issues that we're having with sparks streaming or could they be related to yarn the cluster manager we're using or you know it's possible you know we did not try to get it to try to run sparks cream without the cluster without yarn so it's very well possible that stand alone or on mezzos would work better but I don't know okay any more questions happy to take them offline thank you everyone