scala.bythebay.io: Dustin Lyons and Zack Loebel-Begelman, Akka for Large Data Processing

scala.bythebay.io: Dustin Lyons and Zack Loebel-Begelman, Akka for Large Data Processing

Recording: scala.bythebay.io: Dustin Lyons and Zack Loebel-Begelman, Akka for Large Data Processing

my name is Zach Lobo Begelman and I'm Dustin Lyons and we work with Credit Karma credit call me if you don't know what we do we give free credit scores for around 60 million members today which is pretty cool and this talk is for people out there who built some data pipelines using Scala and maybe as your data is scaled you found some issues maybe we with throughput or you found some bugs as you scale your data and so this is sort of some of the learnings we've had dealing with that so what are we going to talk about today we're gonna start off with a brief history of the data architecture at Credit Karma we'll talk about how has that data scaled we ran into some problems we'll talk about the technologies that we chose to resolve some of those problems how akka streams help us and then we'll also talk about some of the takeaways the results in the learnings of kind of what we learned through this whole process okay so sometime in 2014 this is what Credit Karma data looked like it wasn't the scale we are at today but it was something like this so we were dealing with a lot of analytics from browsers and phones and information about how people are engaged with our products you think a lot of events coming in over real time and all this data is very soon my structured data so this would go into Kafka and then we ultimately needed some of that data to move into our data warehouse so if you're not really familiar with Kafka Kafka is essentially just a high-throughput message queue it's built for real-time streams so we had started to use Kafka we love that and we needed some application to move this data into our data warehouse so we built Scala acha application essentially to import this data subscribe to these topics in Kafka and send this data over to our data warehouse which at the time was Vertica so 2014 fairly simple it out looked in 2015 more people at the company we need more tools to help us look at the data we're collecting so we actually partner with the company amplitude and at the time we had much more data coming through and so we needed some application some service to help us import this data from amplitude so we had some success using Scala naka importing day and Vertica and so we built another service Scala naka to essentially connect to amplitudes rest api import the data from there stick that as well in Chicago and so you know our experience we built some actor systems here but ultimately we moved to ocker streams and that's what we want to talk about today sort of our experience with these two applications why ultimately we landed with akka streams and what that gave us so let's look at the scale the data yeah oh yeah no PII here this is no financial information no credit information this is purely just anonymous data we're using in amplitude just understanding how people are engaged with our products cool so let's talk about the scale of the data we introduced Kafka about when I started at Credit Karma in 2014 at that point we were pushing approximately 162 megabytes every minute and so we had grandiose expectations at the time but the organization was still just getting use to semi structured data in the pipeline and leveraging Kafka so it's not a huge number but it's certainly not a tiny number so the next year in 2015 that number doubled up to 324 megabytes every minute we actually expected this this is what we wanted to see and this is why we wanted to use the JVM use Scala or Java anywhere we touch the data pipeline is because we want to be able to leverage that concurrency it's also why we chose to use auto actors we knew they would scale very well see akka actors leverage something called the actor model the idea is actors are like a really heavy weight function you send them a message think of those as like the arguments of the function it does something and sends another message or does some side effect the idea is that because it's the only interaction here is those messages it scales really well across threads or even actually across boxes you get all this out of the box it's kind of a different way to think about concurrent programming it's a push based model so can anyone guess what our data rates this year are I'll give you a hint it's another 2x increase so we're up to six hundred thirty three megabytes every minute again not not a huge number but it's certainly non-trivial it's not a Google or LinkedIn but it's enough for us to have to do some real work over and make sure without whatever we have scales along with it and so I want to call out this number here 750,000 events per minute that's kind of our baseline rate of events coming in that if we don't keep up with that number we'll fall behind and we'll never catch up so let's take a look at the actor system that we built to initially handle some of this so you have coffee at the top and we have a set we have an extractor or a reader D duplicator and processor so I want to take a second to talk about the constraints for this problem Kafka is a durable message queue it provides at least once semantics that means that you're gonna see every message at least once some of them you're going to see more than that so you have to handle that Kafka also handles somewhat we had at handling semi structured data that it was JSON unfortunately Vertica is a strictly structured columnar store sequal comply sequel acid compliant so we had to convert that semi structured JSON into a format Vertica understood the third constraint is that Vertica likes batch inserts in the hundreds of thousands or even millions Kafka doesn't quite do that that's a streaming platform one event at a time so those are the constraints to your problem and actually if you come in to view our Credit Karma I'll probably ask you to build a system similar to this so let's talk about what everything does the reader just accesses the Kafka iterator gets a message passes it on to the D duplicator that determines whether the message is a duplicate or not if it's not a duplicate sends it along to the processor the processor takes that JSON and converts it into a TSV that TSV is then sent off to the extractor which writes to disk and if there's enough items on disk loads that into Vertica and starts the whole process over again so it's a nice little flow we have here the idea being we have one extractor which spins up a number of each of these actors for every topic in Kafka which equates to a table in Vertica so if we had a new topic come online just a config change BAM we have another set of actors it scaled really well and worked for for our key use case very well for quite a while it worked so well we decided to use the actor system for another use case that was coming about yeah so Credit Karma at the time we really enjoyed using the actor model in akka you know as engineers we could write fairly distributed code without worrying about all the multi-threaded work happening underneath and so Zack said we did have some success with his application and so we were looking as we scaled our data and we're working with amplitude we also use the actor model there as well so just very briefly I'll walk through what this application is essentially these actors are the coordinator actor to pull the data out on amplitude they have a REST API and every hour they make the last hours data available so we had something to kick off this batch process and then we had what akka provides is a router of actors essentially allowing you to scale out work between one actors or a set number of actors you could you could describe so we had a few routers of actors essentially one set transforming the data from the schema that AmpliTube provide and provided into the schema that we needed internally and then we also had a set of import workers to send this data into our HTTP ingest server so at the time you know we're very happy auch is great we shipped this we were happy things were great and then we started seeing problems in production so I was getting paged at nights waking up fixing issues it wasn't that fun so let's look at just a few of the issues we started seeing so one of the main things we saw were futures were timing out and when we dug into the logs and dug into the problems we found that we had sort of a core issue at hand there's only so many connections available to go outbound over I HTTP service so as work was being queued up and pushed through our actor system we work very quickly saturating the connection pool and so all this work is waiting in memory queuing up to go outbound trying to latch on to a future that we use at the time we were using spray and there's only so many connections available so these futures basically timed out as work couldn't be sent outbound so really to get this thing to work overnight we had to just basically increase this time out pretty pretty substantially so that was an ideal but it worked for the time being and so when we look back at this actor system we built you know there's some core things that we noticed so you have this really fast thing happening in memory right so you have this transformations very quickly happening in memory it's happening very fast and then you have sort of a slower thing happening which is using the network and going outbound and sending some data there and then at the time our HTTP NGS server was actually writing files to disk and this was extremely slow right so we've since moved to talking to Kafka directly but at the time we were writing the disk and you know this was slow so we see this problem of very fast work pushing down to very slow actors and slow work so this was an ideal and so another problem we saw actually was you know we can spike up the timeout right we can have things cue up a memory and wait to go out bound to our HTTP server but at any given hour we're not really sure how much data amplitude is gonna provide you know some hours we'd see some burst of data we'd have much much more data from amplitude analytics from our browsers our phones that we get from that hour and so even though that the timeout was fine we'd have so much work queuing up and memory that we essentially just the heap space would blow up so you know trying to tune this in production just to get this thing to run we ended up having to just increase the heap size the max heap space quite dramatically which wasn't that great for other services running on the same box you know because we're just eating up all these resources for work we didn't really need to do so you know I'm running into some scaling problems here and I'm seeing I fundamentally got something wrong and then around the same time Zac also started seeing some issues with scale the data so behind us you can see a chart of this is my application reads from Kafka in the millions on the left and minutes since the application starts on the bottom so we as you can see this isn't great our 750,000 line is called out there and most of the time you can see word literally not keeping up with the data so I had people in my desk asking what's wrong where's my data and I'm trying to figure it out and I realized we had had a new schema or topic come online previously our oldest topic was something on the order of eight thousand events every minute this new schema was eighty thousand events every minute so an order of magnitude larger and clearly the system couldn't quite handle it I noticed it was paging it almost seemed to freeze it was using swap space I wasn't sure what the problem was many of you have worked on the JVM you might already know what this problem is this is halting behavior there's even a 20 minute period we don't have any reads from Kafka I'm banging my head against a wall I have people at my desk I like I can't I don't know what to do and so it was about the top this time earlier this year August dreams had just gone GA and we thought hey this might actually be worth a look so we we took a look at it yeah so Zach and I were very tired of getting paged no one likes to get paged so when we started doing some research and seeing some things that we got wrong we you know we discovered and I think at the time akka streams was in experimental we kind of looked at it briefly but we fundamentally started looking at aqua streams to help us solve some these problems so what is aqua streams so earlier we're talking about we're having these actors push work to other actors and aqua streams flips that it's this idea of pulling work through your system so aqua streams introduced introduces this idea of demand where it's an asynchronous non-blocking signal but it's a demand a message to other actors to say hey I'm ready for work so in my case where I had slow actors it specifically HTTP work going outbound as connections were available in my connection pool it could request more work to be sent so this formal concept is called back pressure this is something we had ignored in our first iteration of our services and just to note aqua streams are still an actor system underneath so you're still leveraging akka and ARCA actors it's just an API on top of that to introduce demand into your system instead of pushing a lot of work and overwhelming the other actors in your system so this is built for high throughput and it helps to prevent some bottlenecks that you would encounter without any concept of back pressure so fundamentally this was something we realized we had missed and akka streams really helped to flip that paradigm for how we are doing our work so let's take a look at some of the fundamental concepts here the core idea of a stream is it's something called a one wall graph and it always starts with a source the idea is that's kind of where the data comes from so as far as the stream is concerned it has zero inputs and one output the idea being that the data comes from somewhere external to the stream once you have a source you plug that into a flow idiom flow is a processing stage one input one output you you can make more advanced stages you can have multiple inputs multiple outputs that's a I think a much deeper topic than we have time to get into but you can have a one-to-one flow and it actually provides a lot of nice things throughout this whole stream you actually know the types all the way through something that with raw alka you don't have so you can have these compiled time checks to say hey this was a sequence of j values but i need it to be a byte string going over the wire or whatever it might be you just introduce a map to make those changes that's a simple processing stage and it always ends with a sync the idea is there's one input and 0 outputs it's the final resting place for your data as far as the stream is concerned right is that it's it does something external to the stream it writes to a file or it creates a sequence that can be consumed from code elsewhere external to the stream so let's actually take a look at some of the specific API that they provide available and all this is code you can use out-of-the-box some of the built-in sources you have from future you have a file in iterator actor F or even a sequence whatever works for you this is all custom code that did you get out of the box these are API is that they just provide same thing with a lot of the the sinks you have head sequence last for each some of these should start to look familiar if you've worked with the Scala collections which I assume we're at a scholar conference all of us have you have fold and you can actually output to a queue and I want to take a second to call out this actor subscriber an actor publisher you have two ways you can integrate with actors if you're converting an existing actor system you can just send raw actor messages in which case you don't necessarily get back pressure or you have these traits you can inherit from that was one of the things I use I use the actor subscriber trait and that way I got back pressure by just leveraging that trait it's define how many how you want to handle items how you want to handle your back pressure and what type is exposed what type the actor and it integrates with let's take a look at some of the processing stages again a lot of the collections API is it's really actually all I think collections API is you have map drop intake drop while filter you know it's this these are all things that you can just use no custom code necessary you just call it on the stream and it'll do that on items moving through there's also these back pressure aware stages this is where you know you're going to have something fast talking to something slower and you want to handle that back pressure explicitly so something like a buffer stage for example you can drop the messages you can fail the whole stream or you can begin to back pressure the rest of it but it forces you to kind of think about how do I want to handle this what makes sense for my use case so let's talk let's take a look at the system I had in how we had to convert it so we had the reason I think we ran into trouble is we had a decently fast operation over the network reading from Kafka we have these two in memory operations that are also very fast and then we write to disk and then we upload to Vertica so we'd actually created our own little version of back pressure because there's only one message in-flight at a time throughout this whole flow nonetheless we still wind up running into trouble we'd actually were creating a case class each time and wrapping an object in a case class passing it along so every message we got from Kafka we mined up making five case classes for we so ended up hitting into trouble running into trouble let's actually take a look at how much simpler it is when we switch it to streams we get rid of most of those actors and there you go it's we just have the stream and so it's much simpler to reason about we wind up having compile-time checks and actually the throughput improved significantly let's take a look at the code and so what was a fairly complicated actor system before is now ten lines of code that all plugs in and can be compiled and you can be assured of the type correctness at that point so we start at the top we create a message parser we mostly just converted that actor says that actor into a object that we leverage here in a map but we provide the schema for the Vertica topic you can see we're creating a runnable graph we completely threw out the reader because we could just use a source from iterator Kafka provides an iterator you can just use we turn that into a type we understand it's a map Kafka message with metadata we did have to write our own custom graphs age that's what that via means just because the D duplicator functionality didn't exist within orchestras so that was really the only custom code we had to write here outside of converting some of our existing actors and then again we just actually map it we use our processor if you notice processor process schema we turn that into a write entry metadata and then we send it off to our vertical writer which we had like I said use that actor subscriber trait which means that this had back pressure all the way through so that created our stream that was our blueprint but it actually nothing happens until we got our start extraction call and we call run on it at that point resources are allocated and the streams materialized and that's when the work actually starts in the stream gets going so this was a huge improvement so much so Dustin actually decided to do the same thing shocking we know yeah in my case you know I already talked about some of the fast and slow actors and how I had no back pressure out the system but there's also a lot of some boilerplate code right where we're having you know a lot of work segregated across a lot of actors so for me the big win was you know when I was able to consolidate some of that work into a stream you know it's just much less code much easier to reason about so let's look at some of that so in my case you know that if the file source is the the start of my stream and so I'm I'm dealing with a lot of files from amplitude and these files you know I think we're yeah the gzip files I'm trying to understand how much data is today but probably in the order of 40 to 50 million events an hour and so yeah simple file source to start the stream off and then you'll see that via stage it looks a little confusing but it's actually just using the new line to break break the byte stream out into lines and then essentially we map the object of something we know about which is the J object and then that transform stage you see there you know before there was a it was a lot of work and a lot of boilerplate and lines of code just to simply do that transform it just comes into one line which is great and then finally you see this map async that's one of the backpressure aware stages act mentioned so it essentially allows me to say hey use all the connections available and you see that first bit the the count there that I include yeah you'll see that and basically that allows us to say when we filled that many connections backpressure so it allows us to tune how many connections we want to have going on at any given time and have the stream provide some back pressure out of the box from that which is really great and then finally at the very end you see we just do a wait for the next patch so all the boiler all the boilerplate clothes boilerplate cloak code excuse me that we had is very simple and easy to reason about under 20 lines which is fantastic so I did mention that we had a lot of heap space to get this thing to work so this is what it looked like before which is not a pretty graph so simply you see a lot of work queue up in memory the top line is the max heap space and then very quickly we fill up memory as we're waiting for you know this work to go out to our HTTP service we probably could have done something interactive system to account for this but I think at the time we started looking at actor streams or acha streams and so you'll see all this work queue up and again it's not always easy to understand how much data we're getting at any given hour so this is one hour but other hours you may have seen this just exceed the heap space so you see it's a very large number of 23 gigs of heap just to get this thing to run so after we switch talk of streams you got much better so here's about a third less heap being used but you see the pattern is much more predictable right we're pretty sistent we're pretty steady and this is just showing the backpressure and what it did out-of-the-box for us just having this data and the rate of work be consistent instead of these huge spikes just makes it much more predictable and much easier to work with so that was a huge win for us fantastic and so I was I was still haunted by this graph right we weren't keeping up it was this halting behavior it spiked up and then do nothing it took a few benchmarking runs but I think I figured out what the problem was but first let's actually take a look at this is our before eyes let's let's see how it does with streams significantly better right here's our 750 line it's not even comparable we spiked up we started about 5 million and then we spiked up and we're consistently between 8 and 11 million so you can see our interest rates have continued to go up luckily we switched the streams and we can keep up now but this is a much better graph this is consistently reading from Kafka pretty much as fast as it can and people weren't at my desk things were much better I was still bugged by that initial that initial graph because I couldn't figure out what it was maybe you guys have and we have a chart up here to show what the cost kind of was so I'm gonna take a quick pause here to talk about garbage collection right it's the reason we all used the JVM is that it's doesn't you don't need to allocate your own memory it meant it's managed for you so we were using the g1 garbage collector which has two styles of garbage collection new gen and old gen cleaning up new objects first cleaning up old objects now both of these are actually soft the world events but our old preocupes is at the top and our post on streams is at the bottom so you can see at the top we're doing these old gen collections which are stop the world events that means halt the application do some pointer math count all the objects how many references there are to them and then clean up the ones that aren't in use we're doing old gen collections on the order of what is that 20 to 25 seconds that's that halting behavior I was seeing in fact there's quite a few times on here where we're doing more garbage collection than we are actual work I think it was creating all those case classes I'm not quite sure but whatever the case it's the switch to streams order of magnitude better we don't even have old Gen elections we're only doing new gen collections and words on the order of 40 to 80 milliseconds several orders of magnitude better no more halting and the throughput as you saw before much more consistent so people aren't at my desk anymore Dustin can sleep through the night I also can sleep through the night and that's my pajamas and things were much better so what are kind of the learnings and takeaways that we have here Rock Alka actors are so very powerful and come with a lot of tools you can leverage but there's no back pressure and so if you have fast actors going to slow actors you can wind up getting into trouble especially as you start scaling up more and more that's when it can really become apparent yeah and so akka streams you know is really built optimized for high throughput you know we still use a lot of just raw actors at Credit Karma for low latency issues so we have a lot of services that run on actors fantastic but if you're looking to solve really high throughput problems you want to look at akka streams primarily for the back pressure you get and again like I said earlier it's still built on top of actors these are still some messages that are being sent but it's just that nice clean API on top of that on top of actors they give you the back pressure and finally don't be like us and try to build a high throughput system without this you'll just start building akka streams yourself so thank you [Music] [Applause]