Scale By The Bay 2018: Jie Ren, Greg Methvin, Using Akka Streams for Web scale Data Ingestion
you hi first I want to thank everyone for braving the smoke and then coming to talk here to talk about web-scale data ingestion so my name is Jay and this is Greg and we're both engineers on the big fast data team at iterable so our team's responsible for designing and building the data infrastructure for the entire company and we're the team that handles all the scale problems that you know a lot of you probably have experienced or maybe want to experience so today I'll start by talking about what interval is and the scale the data that we have just a set of background and then I'll talk about the actual problem that this talk tries to address is and then also why we needed to make this change then I'll talk about the goals that the new solution needs to meet and then finally I'll talk briefly about the tools that we needed to use to help solve these problems and then I'll hand it over to Greg we'll go over our a new architecture and including Scala code that we wrote and then we'll finally do a look back and then open it for questions but in case time is tight Greg and I will both be at the it or good iterable booth outside after lunch so feel free to come say hi and ask questions so what is iterable we can be described as a customer engagement or marketing growth automation platform we're a b2b company and so all of our customers are other consumer companies and our customers use us to send campaigns to their existing users in hopes that their businesses can grow bigger a simple example of this can include hey you like this artist and they're coming to where you live in three months so buy tickets and no business wants there are campaigns to be spam and they want to send information that's relevant to each individual recipient of their messages and you don't want to so for the example above example you don't want to send that reminder to everybody you just want to send it to people who are interested in the artist and who live in the area so iterable provides an automated way for them to do just that they can dissect and segment their user base to figure out who to send what content to our customers can use us to send campaigns email mobile notifications now really any medium that they want and iterable enables this at incredible scale because we have many customers from a variety of different industries and all of these businesses have varying numbers of users so here um 18 of our customers that I actually stole from our sales slide deck but I did want to use it to give a glimpse at the type of the scale that we deal with on a daily basis so if we look at just Spotify they report having 83 million paying customers as of this year but you can imagine how many actual users that they have and in total including the ones that don't pay and we don't only have Spotify data we have door - Zillow box seat geek 18 tea all of their marketing user data so in all we have over 250 customers and we primarily hold two types of data the first one is user profiles so our customers customers information and then events which describe the individual actions that effect each user so did iterable send them something did they make a purchase did they subscribe to some list or something and so the data comes to us peak rates are on the order of tens of thousands per second but we see this fairly regularly but it averages to be on the orders of thousands per second it ends up being over a hundred million I did new documents added we only see these numbers growing in the future because as our customers grow they use us to grow so they will grow and we're going to acquire more customers so at the end of the day all the data lives inside elasticsearch wheel because we want to use it to allow our customers who are marketers to filter and aggregate and make sense of the data that they have with us so these numbers I pulled a few weeks ago we have over 40 terabytes of data over 60 billion documents around 2,500 elasticsearch indices spread across 200 data nodes so the problem as you've probably gathered is that we have a ton of user data come in so how do we design our data pipeline to be able to handle the ingest process the data in the order that we receive them and then make that data are useful to the businesses that sent them well it turns out interval has been around for six years so we already had a solution but it just didn't do everything that we wanted all the time and all the times really important as we scale so the way this old architecture works marketers use our API to send us requests for events or user updates then we buffer those requests inside RabbitMQ which if you're unfamiliar is a queuing system and then our ingestion service uses akka actors not akka streams to pull messages or sorry it gets pushed the message off the queue and then index into elasticsearch one thing to know that a marketer when they send in the initial requests they can define a custom trigger event that will happen after the original event is ingested so then if those exists we publish them back interrupt thank you and don't get that's the same way so when is a change about this architecture one things we really need a change was RabbitMQ basically in its entirety the way we're using it read and queue right now is a single point of failure if it dies were no longer processing anything until it's brought back up under very heavy load flow control inside rabbit and Kia will block new threads from publishing to it this means we're we would see severe slowdowns and requests timeouts on our API layer and the rapid MQ client that we were using also does it implement back pressure and it doesn't slow down when grab the rabbim queue is overwhelmed like it's supposed to a lot of things are kept a memory and is another reason why rabbitmq dies we also have the hard code a hive prefetch count inside rather than queue to allow for high consumer parallelism the prefetch count indicates how many messages are pushed to a consumer at a time but we should really happen is that the consumer should decide how many messages they can handle and for consumers overwhelmed they shouldn't be like we shouldn't be shoving a ton more work that's throat as it's currently occurring when you do that it just makes our problems worse so with the existing architecture we can't dynamically change the prefetch count and it's just hard to control the concurrency of parallelism within our dreams another point that we really need to rethink was how we were doing batching in the API layer that I showed earlier we expose bulk update endpoints but we don't have any restrictions on the size of a batch or the size of each individual update within the batch so not only that but we include batch requests verbatim into a rabbinic you exactly the way our customers who are marketers set them up so what we end up seeing is batches that require varying amount of work took to process so a gate is not regular at all and we also suffered from messages getting processed out of order in this in the specific way we were doing batching I'll show you example of that so imagine that a customer sends us batch a followed by batch beef and they both update the same field subscribed on the document my document but batch B is suppose we processed after batch a but what happens is that that J will actually can be consumed by one consumer and batch beat by another consumer so when this happens we have we can't guarantee that B will finish after a because we have no idea how long each batch will take to complete but this is really it's really important that we we we get the correct information because something else could be keying off of that value to figure out if we should send someone something and if we screw up the order you're doing something completely different than what the marketer intended so something else that we found was really flood was how we're doing error handling a little bit ironic but when a RabbitMQ consumer encounters or a couple Bellaire and while processing a message it just kind of gives up and then every entire message back into the queue I found it from but there are two problems with this remember that rabbitmq a rabbitmq message could be a batch of requests but maybe only one requested that batch failed but when we rien the entire message what we're J's reach we're retrying all of the requests in the batch so we're doing it including the successful one so we're doing a ton of extra work that we don't need to be doing a second problem so much of the ordering problem from the last slide we could also lose order and cheese when we do this because now the message that's getting retried is actually being processed after messages that came in after it originally so we saw in production where performance metrics that looked like this it was super spiky in this graph even though peak performance is about fifty thousand messages per second due to all these zeros down here where we're doing zero work on average we're only achieving about a tenth of peak performance that's not what we want has anyone ever sat in traffic in LA yes I haven't it sucks but anyways I found an article I said a few years ago the average speed on the 101 on a Wednesday night in rush hour was 19 miles per hour I don't believe that actually I think it's lower but regardless the actress speed limit on this highway is 55 miles per hour and I know that most people drive above the speed limit but ideally in a perfect world there'd be no stopping and there be no stops there we just know stop and go traffic so that's what really kills you kills your overall speed yeah we want that traffic in our dart data pipeline so all of these problems that I mentioned manifest as real-world issues that affect real people our marketers send us messages that you know we're sometimes not as reliable and delivering and engineers like Greg and I get called in the middle of night with alert saying hey keys not processing and it's like figure out why about everything so intertwined but it's it is really hard to figure out and it's also really hard to fix like even if we do have a problem it's usually just little add more nodes and we don't want to constantly be doing that also of course documents are occasionally updated out of order so this is just a correctness issue that we're not doing the correct thing once we're because people use that data so given all the problems we've identified we summarized concretely some of the goals we wanted to achieve in our new pipeline that will ensure faster and more correct processing if you remember our highway analogy we want to achieve that constant rate of processing at all times for maximizing throughput we want our speed limit to be the speed at which elasticsearch can write and nothing that comes before it just slow us down we also need to ensure that API requests always work that we can always hand our customers with it no matter how big it is so we need a queueing system that's distributed and highly available to buffer these requests we also need to ensure that updates are always performed in the order that we see them so that the data is correct and finally we need to handle errors more effectively in order to achieve all the above goals but it's important enough that it gets a bullet of its own so here's some of the tools that we're going to use for in our new solution and some of them will just solve some of our goals right off the bat but other things require more things so the first one is Kafka or it's a distributive publish/subscribe system for storing and processing streams of data and we chose it to replace Rab MQ it's highly available and we should always be able to write to it it allows this this will allow us to handle a spiky load much better it also lets us define the number of partitions and therefore determine the amount of parallelism that we want so that we can maximize throughput another tool that we use is akka streams which provides really nice high level api to slice and dice our data streams allows engineers to compose a graph of many steps that the data flows through augur streams implements reactive streams you guys are on the reactive room and this makes sure that back pressure is propagated to the source of the stream this means that no stage should ever be overwhelmed with more work that can handle it gives tools to determine how data should be batched and allows developers to easily determine the amount of parallelism that's needed at each stage in their stream and as Greg go show it later it's easy for any developer to see exactly what's happening inside the stream at high level just by reading the code and then we also use the our pocket cut Kafka connector which turns messages from Kafka as a source for the our streams consume so and then I mentioned in elastic search a few times it's a distributed search engine both naba leucine and interval uses it for its fast aggregation and querying capabilities so that marketers can tell exactly what's going on with their users and what their events it supports no sequel what just works really well for us and it's actually really important for us because we have no idea what format our customer's data comes in as and every customers data is different than another customer so we don't want to set limitations for what kinds of metrics that they want to track by using else and now I'll hand it over to Greg hook over the new architecture and then show you guys some good so--that's thanks Jay today I'm gonna go over our new architecture so here's just a reminder of what the old ingestion architecture looked like and here's overview of what our new architecture looks like so what's different here well first of all we're using Kafka at around MQ and this really solved our flow control issues with rabbim queue right off the bat and then we have a tonka streams graph that consumes from cough cough comfort the Kafka partitions that are right in to elasticsearch unlike our previous architecture we decided we wanted to detach all the bulk API requests completely when we publish the Kafka and we want to debauch so we've out of work for each individual individual message is more constant and the Kafka client will when you actually published the Kafka client will do the batching automatically over the wire so we'll still get the benefit of batching over the wire but we'll see we'll see each other individually in the Kafka topic when we consume from it and we also use Redis here to deduplicate this gets us a little bit closer to exactly once even though Kafka at least once if we if the consumer were to die and we need to restart or if we retry a publish we could end up with multiple messages so we use Redis to check to see if we've seen a message before so at a high level our graph is parsing the request then it will rebadge then it will index those batches in elastic search then it will trigger any subsequent actions which could things like sending emails and triggering new events and it commits our Kafka offset and the back pressure of any stage would propagate it and eventually manifest this Kafka lag so if any of any of these stages are too slow we'll just see some lag buildup in Kafka and so for our petitioning strategy we wanted to guarantee the updates and events for a single user or process in order that was a very important goal so we can do that by partition by user ID this way updates and events can be processed by the same partition exclusively and put in by by the same stream and we're have we had we'd have a single thread that's processing third of execution that's processing each partition and we can add partitions if we want to increase parallelism this worked really well because we have a large number of users and the events and updates are typically spread out among a lot of users so we can get a lot of benefits from doing this partitioning strategy and we still can get inorder updates so I'm now I'm going to go over sort of a simplified version of the code that we use to implement this logic and aqus dreams it's pretty easy to understand what's going on once you understand the basic akka streams primitives first we use the committable partition source from the Al pocket Kafka connector and that gives us a source of sources one source per partition so when Kafka rebalances partitions those sources will complete on the consumer that's it's revoking that partition from and it'll restart on the consumer that it's giving that partition to and then map basic and ordered will run all these sub screens concurrently and we just omit an element downstream on the the outer graph when each one completes and this just allows us to track the completion of those of those sub streams so now I'm going to talk about the how we process the stream for each individual partition so the first step here is to parse invalidate the message the messages all will enter that stage is a byte string and the leave is a validator message for the next stage to process so as you can see we got some zeros and ones that come in and then we this nice pretty validated document and here's the code for that stage you can see it uses map basing because we wanted we wanted this well the actual logic here would usually involve a fetch in a cache or a Postgres DB that that is asynchronous and the parallelism here we can really use any parallelism we want because the ordering doesn't doesn't really matter here the next we want to batch these messages for indexing into elastic search and we use a really nice akka streams helper called group within which allows us to bet to batch elements based on a batch size in a timeout so for example if we set group within to be size four with a timeout of hundred milliseconds and six missed messages come in as a burst and we don't see any message for a few seconds then we get it get out a batch of four and then the other two would go into another batch that would be sent off after 100 milliseconds and the code for that is pretty simple and these numbers are things you could tweak depending on how what the optimal thing is for for your use case for us it was around a thousand or so we wanted to get for efficient indexing into elastic search this will admit batches to the next stage so the next stage is responsible for actually indexing those batches into elastic search it's really the most interesting part of the processing logic and there's a couple of things that aren't really shown on the code that I want to go over here when a batch enters this stage we first we want to check Redis and filter out any already processed messages this this as we discussed before this could be if we had a duplicate published or we had the consumer restarted in middle of processing and we needed to we saw messages we already had we didn't commit before this just makes it less likely that will will accidentally process the same message twice next we want to merge updates for the same user and this is sort of an elastic search specific thing elastic search will perform really badly if we try to update the same user twice in succession it could take like a couple seconds because it has to do a force refresh so we want to avoid that for free fresh and actually if we have a bunch of objects for different for the same user we merge those together into a single update the next we want to bulk index into elastic search and track any of those index messages in our Redis consumed element tracker and finally we want to retry if we get any retrial errors so if we get a 429 from elastic search meaning basically elastic search is too busy doing other things if we get a 409 which means as a conflict we can retry those and basically retry with a back off until until it succeeds and finally we will omit those index documents to the next stage so here's how we add that to the stream all that logic is in this do dupe in index function it returns a future and again we want to do it map based ink with parallelism one so we can maintain a strict ordering here this is just per per partition so now we want to trigger any changes that would happen as a result of indexing this document for example we might want to send an email to the user after they they get a sign up event so those triggered events there may be triggered events that would publish back into Kafka for later processing by the addition pipeline and after we've done all this we want to omit the ending offset of the batch to the next stage and here's the code for this and I should also mention that because we've used this Redis tracker to filter out process messages this is basically at most once meaning that in rare cases we might have a failure that would cause us to not send an email but for our use cases we decided this is better than if we sent ten emails accidentally to the same user because we were restarting something and we have will have ways of tracking when this happens so we can we can figure out if we forgot to send send messages later in the process so the last step is pretty simple we just want to commit that offset in the bash to Kafka and we emit a done we also use a retry here because occasionally in Kafka the Kafka might be down that Kafka broker might be down for a short period of time a couple hundred milliseconds and it might make this fail so we want to retry so we actually succeed at this so there's the code and under the hood this will use commit Scala DSL from the alpaca API and it's just just a wrapper that wraps it with a retry helper and again we want to use a parallelism of one because we want to make sure that we commit these offsets exactly in order and finally we use a nice helper from alpaca called draining control which makes it really easy to shut down the stream there's an example of this in the alpaca documentation as well so we use drain and shutdown on this this stranding control and we just need to wait for this future to complete and that will trigger the completion of this outer source which will complete the inner sub sources and then those will admit elements downstream and we know once we've emitted those elements downstream which are just akka done elements we know that the sources the petition sources are all done and we can safely shut down our application so error handling was kind of an important thing we had to deal with so how did we think about that so in our case we had a lot of unrecoverable errors because we our customers kind of define their own data model so we might have situations where we didn't do a good job validating that data and or they sent us something that we didn't expect so for those we typically just track the errors and we continue this dream and that just that made sense for our particular use cases it might not make sense for everybody's use case for recoverable errors like conflicts vs rate limits some of some of which I discussed before Kafka Broker failures we could retry those with an exponential back-off and for there's also a whole class of errors we may not have actually foreseen before we want to in that case we just we actually just do the default which is to stop the stream innaka streams and that's because we prefer to err on the side of having somebody manually figure out what the error was and handle it rather than trying to guess what to do and potentially get things into it into a state that we didn't expect and we also use a custom akka streams materializer if you're not familiar with akka streams which of the materializer is the component that actually runs your stream and we add a logging like this basically a log there that logs whatever the error is and just by default the exception might be swallowed in certain cases so this is just generally a good idea to make sure that your your keep tracking all the exceptions that happen it's easy to see what's going on so I've tried to summarize our progress in some pictures so our previous architecture it worked but it was kind of like a dumpster fire well the dumpsters happy because it usually worked but there are a lot of situations where we just catch on fire and not do what we expected it to do and I kind of think of our New York Fisher like ice cream GA really likes ice cream and it ice cream makes her happy just like the new architecture because she doesn't have to wake up in the middle of the night to deal with things going down and I think pretty much everybody likes ice cream right let's relax intolerant I guess and it just handles scale a lot better it's a lot easier to maintain and our customers were were much happier with it and especially you aqua stream has just made just made the process of developing it a lot easier yeah so that's about it if you have any questions you can ask them now if there's still some time but you can also come find us at our booth at the iterable booth yeah thanks for listening you