scale.bythebay.io: Evan Chan, Building a High-Performance Database with Scala, Akka, and Spark

Recording: scale.bythebay.io: Evan Chan, Building a High-Performance Database with Scala, Akka, and Spark

I'm not sure there's like an official start signal how's everyone doing good not quite awake yet awesome so today we'll talk about building a high performance database with Scala akka and SPARC based on you know projects that I've been working on my name is Evan Chan if he forgets my name is on my last name is on my shirt you know so you it's only four letters so it's not hard to four to remember but I've been using Scala for a long time just a show of hands how many people are coming came to Scala from say Java no it's like a lot of hands how about Ruby few hands so I came to Scala from Ruby and it came to Ruby from Python and if you come to scallop from from Ruby is actually very easy because Ruby is already a semi functional language well you've got a lot of debate in this room about that so I won't go there and I have been user and contributed to various Big Data projects including Apache spark for a long time a youth spark when it was 0.8 way back like before mostly were heard of it and it was basically like I just evolved from a research project out of Berkeley and the configuration of spark was done by setting system properties so that those were the good things and I've done talks at various places the spark submit Cassandra submit and I've linked there you're interested in looking at blog posts and other stuff like that so today we'll talk a bit about the debate database that built follow DB but a lot of this talk is more the theme is really more about how do we use the abstractions in Scala to build a big data system people do that's a lot a lot of us might not be building databases but a lot of us would be like you don't want we want to know what like what parts of Scala to apply and there's a lot of good lessons but I'll start out with white belt a streaming database so nowadays like we're dealing with huge volumes of data like huge volumes of streams right you're talking about like there's a lot of IOT applications there's a few talks here about that I noticed that are about LTE which is kind of cool but you can imagine like yeah IOT you know operational metrics like no matter where you look there's like huge streams of data and oftentimes nowadays it's really important to deal with real-time and low latency data the two concepts are slightly different but essentially you want to get at the data that you're ingesting right away and to answer questions increasingly right away for example dashboards right well that's kind of useless if you get data and you bash it up and then you serve it like you know oftentimes like if it's an hour later it might be too late right and oftentimes we're dealing with flexible schemas you know data might change and in all these things we want systems to be as simple as possible if if you've production eyes and debugged streaming systems you know like those are often times the harder systems to debug because of the velocity that data is moving is not like I run a batch job one time and then loop something went around let me run it again and it might be okay but in streaming it's not okay you can't even have your three systems be down like even one minute because you could be losing a lot there right so a general architecture for this stuff kind of looks like this right you have events coming in to some your message queue and you have some stream processing layer and and then you have state and and a really interesting question I guess for us is what does the state look like and hopefully like out of all this you get happy users you know you're serving stuff out of your state a lot of so a pretty standard solution that you know folks use is something like this like you use Kafka is to message queue that needs no explanation may be use for extremum to process it and then you kind of write small files now why do I write many small files if you spark then you know the spark streaming at some micro batch architecture so it takes data and small batches like one batch one batch one batch if you want your thinks through a real time you can't spend too long writing these files like maybe I don't know let's say that you want a 10-second window right so now you're writing files that are 10 seconds long so 10 seconds files are not really that big so what you see is you end up writing a lot of small files and it turns out that you know HDFS doesn't really like having like millions of small files so and it's also very inefficient process so you end up having to consolidate them if your data workflow involves data that you might see again you might have to deduplicate so after a while you realize that this becomes a fairly high latency workflow in order to get to a point where you can really like read data efficiently and the fundamental problem is that there's a big impedance mismatch it's like trying to fit a circular circle into us well sorry a square into the round hole right streaming data and files are fundamentally different abstractions right and trying to fit one into the other leads to a lot of pain so you have Apache Cassandra I think all of you guys are probably very familiar with that Cassandra that they distributed key-value store into the database really it's more than key value store you can and it's designed for you to write tons of data into it and you can create data right away so it fits a lot of the characteristics that we are talking about so this all sounds really good at a high level now if you work with data at a high enough velocity and scale and volume you soon discovered that just working with sunder out of the box has its own pitfalls oftentimes you find that the volume that you're able to write using just a native API is it's not quite what you need so you end up having the massage data in order to make it really work for large volumes the other thing is that priests have to be very simple they can be a little bit more than key lookups you can do like range lookups but essentially it works best if you read from one what is called a partition that's actually one note or one thing at a time now you can change that you can add and there's a lot of you know stuff that folks have done I'll point to one of my friends here you know Russ that works at data stacks has done a lot of great work on making smart work with Cassandra so that allows you to do flexible queries but you make some trade-off so you get very flexible craze but then you won't you might not be able to do a lot of them at the same time and they might be higher latency right so out of all these needs is why we wrote fala DB to be a streaming solution that can ingest at a very big volume and also answer low latency queries quickly although I'm not going to be talking that much about the product itself it is open source you can look on the website but the idea is that you can ingest stuff and be able to meet different needs going to dashboards for dashboards so typically like you have a lot of users running quick really quick queries and as far as for ad hoc and other needs which are fit more traditional spark where you need to run like only feel them but there are like much more complex so we'll come back to this but one what I want to focus on though is that we built the stack of phyto to the spark based on a hundred percent a reactive stack so this is all the code is in Scala which is I'm proud of that back and we use the aqua cluster extensively SPARC is part of it important part of it we have increasingly getting into using reactive streams I'll be talking about that a bit and we use futures and part of the talk will be about what what parts are use where we'll use types of config for all configuration and there's other random libraries that I'll point out that I think have been very useful to us even even the highest performance parts and I'm quite proud of the fact the parts that have to run within one nanosecond scout so using Scala and akka and spark for database so first of all why use Scala and akka so I would say that a big part of it is that we really get a lot of the right abstractions and some of these are not all these are built into Scala like akka is you know a library obviously but I think if you combine the whole ecosystem you got a lot of really useful abstractions you got streams parallel reactors dreams you have futures you have you know the actor system you have distribution you have a lot of important components that you can combine and pick it's almost like what X is I'm so big that you know we have to actually pick the you know the right the right ones which is at this point right some of the properties of frameworks like akka are critical for well for all systems but a product for databases by failure handling being able to recover and be resilient all the time and and of course all the pattern mashing and type safety I think is actually quite crucial you never get stuff right the first time so being able to refactor fearlessly is I think one of the best reasons to use golly like when I have to refactor stuff I don't really think about oh you know like a like in Ruby am I gonna get a method not found error when I run a job and like two hours later and into a Hadoop job I might get this like funny but once upon a time actually had to run Hadoop jobs in Ruby so anyway there's a there's quite a few Scala Big Data projects like obviously I'm where no spark you guys might not all know geo Mesa it's a really really interesting project out of the Eclipse Foundation I forgot exactly produces it but basically it's a Geo it's for geo data and you can run geo queries on it which is really really neat someone's written I think these came out of Comcast I think there's an akka time series database got Cronus there a distributed key-value store called serious oh sorry serious comes out of Comcast I think so so three big abstractions actors futures and observables are what we use for reactive streams and so when when you use it in what parts so here's what one fellow do we know it might look like at the very top we have a couple actors and you can think of that the data that comes in in the system as well as commands and user commands come in through a single coordinator actor at a top to chain so all everything comes in through there and gets routed to different things you can think of there's their underlying actors that represent that handle ingestion for one particular what we call it dataset so just like in SPARC you have different datasets are basically different pieces of data what a with a single schema all right like a table so you might have different actors for handling ingestion for different data sets and we might have actors for handling querying and so forth and below that layer we would have more low-level constructs that are useful for accumulating records in memory organizing them flushing them to disk in a more optimized call dinner fashion and then they get written to a column store so what you see is that we can organize this functionality in different groups at the very top you can see that acha is and actors are used because the top layers have to do a lot of coordination and we would use futures and observables for the lower layers where type safety and back pressure and other things have become more important so use actors for basically external API s for doing a synchronous messaging to and from clients we request response other things like that as well as for cluster and distributed state management so that that that is a pretty good use of vectors in my opinion and then we use futures and observables for core i/o for doing data processing and ingestion maintaining type safety the actor API is not it's basically like you can adjust message of any type so it's a little bit not not as safe and and for doing data flows what backpressure so an example of this is that for actions that are clearly single actions then that Maps very well to futures for example like I have a an example like drop data set it's one API example right it's it's an action that you do one time you get a result if you get an error that that happens when you execute it and you also get an error back from the future so this so this may have to very well to a future because futures run one time and then you get back this response right now in the when we originally wrote this there was actually a lot of the reactive streams landscape was was really like NASA in it was just starting out and and and so we did a lot of things with futures including things that probably should not have I give an example of of an append of a a write API but a lot of these things who have started to migrate into using reactive streams so out of curiosity how many folks are using react to streams in your stuff so few I don't know maybe 10% issue or maybe 5% so you've probably heard a lot about this reactor dreams I think there's other people that there's a talk on aqua streams later the JDK 9 I believe has a flow API now so this is becoming a really hot topic and with good reason well what our streams well everyone probably knows what an iterator is right so an iterator is something that iterates over a sequence of data in a single thread what streams are is that they are asynchronous dreams based on observable sequences and they are asynchronous which means that different parts might execute on on different threads and it is designed from the ground up to have back pressure which is I think a really important part so it's asynchronous yet so you can have several parts that might be executing parallel but at the end of the day all of them actually conform to a back pressure API so you know that when you produce it will a producer will wait for the downstream to be ready and that's built in now monix is a particular reactor dreams library that has been built from the ground up for Scala so it interacts very well with Scala its integration with a lot of things in the sky ecosystem including futures and some other type level projects like cats and other things like that before we were using things like future iterator which really doesn't work that well it's not that's not asynchronous really it just happens to give you an API that we turn something in the future but streams are much better because if you start thinking about things what one example is you know say weeding data right now it might seem to you that if you have a standard app reading data might be something that you think of as a future like basically you get some answer in the future but in the database basically reads have you want things to be streamed because if let's say you're reading you know a a few rows then the public doesn't matter right if you're reading say 100 million rows on the other hand you don't want to wait for all that to be done to get an answer back what what you want is to have a street a streaming API so that you start getting data back and in sets write in parts then then somebody else can work on that like say spar can do some processing on it while you fetch more data and then the back pressure comes in so that if the processing is slow then you don't you know push more data than then the downstream can handle so that in this example we're we turning unobservable and what happens is that each part so scan partitions partitions are kind of like groups of objects in follow DB and you can think of an IOT application might be like say each entity or each I don't know whatever making and so yeah one method that might go through and return partitions and then what we do is we do a flat map and then for each partition we might go to disk and read this from say a cassandra or something and a really nice thing about what reactor streams gives us is that the repartition chunks could actually be is actually asynchronous it could issue parallel weeds to two discs and what monix allows us to do is allows us to merge those results back in a way that still gives us back pressure and and it's very easy to merge them the flat map would turn each partition into into a sequence normally you think of scholar collections flat map takes one object and this all happens in a single thread but in this case the flat map the things that underneath the 5 map could be running synchronously and this mixtures that it all works so so that way we get we can control the parallelism and get you know very pretty fairly good performance and we still get a really nice compact representation we can use scan and collect and do things what the results another area is in terms of ingestion so what we get what monix the first thing you notice is that we have emerged so we're able to merge local streams together one stream is a lot faster than the other in this case and but it still get interleaved correctly what's back pressure and then you notice another thing which is really interesting called map async so what map we think does is it will take objects off of a stream and it can fork them into tasks which run in parallel well in this case non-parallel is what controls the the number of parallel things that are happening so if say it's it's eight or something that means I kind of eight tasks that are right so I can take eight objects from the stream and random in parallel and if one finishes it would take another one off the stream so this is a fairly powerful abstraction you can control the parallelism really well but the whole thing is still back pressure and I know that if the flush is slowed down let's say that oh eight of them slow down then it will stop processing upstream more data that comes in so this is a really important property to have and data flows monix in particular is designed to be like really fast like to the meadow and and give you really fine control over parallelism and other things like that which is one reason why we use it and the other thing that you notice is that there's also error recovery there's a recover and err right okay so I'm gonna move on a little bit to talking about mixing in aqua cluster and spark which is fairly interesting you might wonder why we're doing this I'll explain this in one second but so in spark you have an executor thread sorry a driver thread that runs your application and then you have executors right so where with me for a minute let's say that you have an actor in the driver that can respond to membership events in our cluster so what happens is that when an executor comes up it will send an events to this actor in a driver and it would know about membership changes so it knows that an executor is coming up and in this case executors one comes up and then an executor two will come up and you get another member of events and then the no cluster actor knows that another one has come up now this is pretty interesting and it's pretty non-standard what would we how is this useful so an example is that this for sending commands so this is something that because we have a lot of states in our data processing layer we can do interesting things like tell all of the workers to flush the data so this is something that is like basically it doesn't fit within the SPARC abstract because spark is us for processing data so in this way we can send a skewness commands we can tell each one to flush and in this one we'll flush the data that your memory to to disk why is this important because for certain like let's say that we're interesting whole batch of data and spark at the end of it we might self state remaining in all the executors we want to flush that and ensure that data gets written so that you can read it properly right so so I already talked about basically async messaging that's one in a really interesting use case the other interesting use case of why we integrate our cluster in this park is for stickiness so a spark is not inherent design for sticky operations meaning that when you get each new batch of data let's say I'm sorting it well with each new batch the the range of data that gets sorted to each worker is going to be slightly different because each set of data is different right but for some of the processing that we do for example we encode data into a columnar format you really want the same the same key to go to the same machine again again so you need something else stickiness the implementing stickiness is it's very non-trivial like I've done it several different ways in spark we decided to what we decided to do is to route these incoming records maybe this will be a better explanation so basically you can imagine that each executor will get some sore throats what we do is that we have an actor that can actually cut a sort of role source that can actually take this data and route it to the right and other node and routing it's not the same as sorting sorting means that you actually have to take every record like see a hundred million records you have to sort all of them biased or an order routing just simply means like I'm not sorting it within each bucket but I'm just sending each one to the right destination and the really nice property about routing is that one is that I don't need to wait for sorting to happen so that's saves a huge amount of latency when you when you ingest and it allows me to act on stuff right away so that's so that's another use what we do is that you see there's a no cluster accurate I keep track of that is used for keeping track of the routing assignments so it knows what is the range of input keys and to to route to each thing you know it's actually like a token range and so we can use that to route in real-time to different parts one really nice property about okay so you notice in this diagram everything this is happening it's in one process executors one process but let's say that I want to change my architecture around and maybe separate out part of this to like a different process right well what the brilliant thing about akka is that you can basically do that without changing basically without changing code so I can remove part of this like let's say you want this to be a separate process because it has better performance characteristics is better GC whatever you can do that for for for actors when you send message across it uses what is called an active reference it's a reference which can be local remote from the programming API it doesn't matter so the really nice thing is that you can reconfigure your architecture and your code remains pretty much exactly the same which is really really nice so you have two tests other stuff right how do you test it well Ark has a really nice testing kit called multi JVM and it actually is an SVD plugin it allows you to spin up multiple processes and it has things like barriers so that you can wait until all your test artists are in point together to execute things and you can even test that you can assimilate Network conditions like inject bolts and things like that so it's a really useful really useful test kit there's another library we use which is really cool caught I came on I don't know if anybody's heard of it but k1 is like basically like metrics for the reactive stack it gives you the ability to trace as well as have metrics on actors and things I'll talk about tracing a little bit what what tracing means is that let's say you have something that is broken up into multiple asynchronous segments like this is a right this part of right workflow so that something like right Renison right into this might happen in parallel they're both futures even within right chunks things might you see a future uh Traverse so I might be doing multiple like things in parallel so what the tracing allows you to do is allows you to tell how much time is spent and once in the first part or in one part how much time it's been in another part it well like actually measure how long things take and it doesn't do this for every single invocation because that would be expensive so you can have a sampling rate and you might stamp over say every thousand invocations or whatever and it will dump this data out and allow you to see it which is super useful for debugging and for performance things and it has a metrics system that because that integrates natively with akka actors with akka HCP and other things like that and actually and the histograms use HDR histogram which is a really good library for measuring precise lanes ease for look validation we use something called galactic people have not probably have not heard this much but it is actually built into Scala tests and it allows you to do if you look at this for comprehension this is what we validate schemas like it's there's a lot of different steps to it so you can actually compose this really well where you know if the first step get computer columns fails and it will fail with so the API looks kind of like that and it looks like you gotta we sub back or you know something else which I think it's quite readable so it will give you back this error and it will keep going until it's validated every single thing so it's so it's a really neat library and you can compose these things fairly easily and the last part is hawk I'm going to talk about machine speed Scala this is really neat like how to get things to run really really fast a lot of times like people tend to write these parts in Java but I'll tell you that you can definitely write anything that runs as fast as Java and Scala it's just maybe a little bit more work and won't look like the scholar that you loved necessarily how do you go really fast well you know you avoid you try to avoid some things as much as possible try not to serialize things you try not to allocate things which sounds like impossible and you try not to copy things yeah so are are under our abstractions under the hood are columnar format can can do like you know can do billions of items single-threaded if we go through spark is a little slower because a spark has a has a role abstraction that slows things down a little bit but it's still in the grand scheme of things extremely fast this is the repo underneath that we use you can think of it let's see what's the best way to think of it you can think of it as like a a scholar vector or an or an immutable sequence that you can read back extremely quickly and which does not need to be sterilized basically when you create it it is already in the binary format which means that it can live off heap for you know GC less memory management and you can access it linearly or randomly and it is also very compact because it uses call nor encoding techniques we use gmh for testing performance so this is an example like of the loop that we use for testing it is a for loop notice that there's an optimized keyword there's a special Scala library called galaxy it might have be renamed now but it basically does the macro that we write before the into a while do so what we're doing is a really tight loop that all were doing is pulling one element at a time and summing it and and so this runs at a two billion ads a second in a single thread right so who said Scala cannot be fast and a lot of magic for getting this kind of performance is using something called JVM inlining and basically the trick is to have a lot of really really small methods so at the top the apply method gets out an individual value let's say that this is using differential encoding so it might add something to have base value underneath the hood another layer we might have another thing like really small method that basically you have a different method for different types of number of bits for example like there's one for 32 bits there's another one works with bit values you know and everything gets converted to an at the end and at the lowest level you have something that works at a you know at a very low level to get out the actual ends from the machine and this is using some misc unsafe which is basically translates directly to a machine read operation so the JVM will like as long as it is under a certain number of bytes or sorry not bytes but yeah certain number of JVM off codes basically the JVM can inline these method calls into a single runtime which makes it really really fast and but still gives you a way to not copy and paste much code you can still have some nice abstractions doing a final death is very important because that avoids virtual method dispatch that lets the GBM know that there's really only one instance of this method and so it that gives it further optimization so it's still possible to use traits with what this really high-performance thing the other thing the other example of doing machine sees well I'll go into is something that we call a binary record so in most applications you actually know what your domain classes are like you would model them as case classes in in a database we have a tougher problem in that we have to support flexible schema so at runtime someone can define like say tomorrow you know someone hasn't has a new type of data they want to ingest right so I would have to define that dynamically right I would say you know this type of data has like events ID field and it has a timestamp view it and it has like you know these are ten fields that are domain-specific but they don't know that ahead of time right so we have to deal with flexible schemas but we still need a record type something that can hold like say one record that has these different fields and we want it to be fast so this is a fairly difficult problem right like you can think of what generic ways can I do it as easily in Scala well there isn't really I mean I can use something like a seek of any for example right there so that's kind of really slow you know it means that everything has to be boxed into an object and in serialization becomes very costly so the solution that we have is something called a binary record what it is is it's essentially a binary data type that means the data is already sterilized when it's created and it's but it supports different schemas and if you can extract values with no serialization and the way you do that is that part of the trick is that we have a special string class which is a utf-8 encoded string class it is basically translates its strings that are already sterilized in the JVM Java strings are actually all 16-bit characters so this also saves a lot of space over the wire and once you build a record becomes immutable this sounds a little abstract but I think we look at sorting I think we've become a clear why why this is a huge benefit so if you have sort something basically let's say that you have a standard you know like protobuf record right what you need to do is first of all you got a protobuf you need to you serialize it into an object the once its deserialized you have to create a sort key so that might be another step and at each step you have to do an allocation you have to do some of the facilitation and some work then once you have a sort key in memory then you can actually compare it and and so this might seem like something that is like an obvious way to do things but when you look at in depth you realize just how expensive it is there's a lot of memory that gets allocated there's like an incredible amount of code that has to be actually run to even make this happen with something like my new record we know if we don't ever just utilize or allocate memory what we do is other than actually allocating for the actual binding record itself like from the network or whatever what we can do is for example that you have a name field and you have a timestamp field we just compare those directly in memory and we don't reallocate anything so that's the kind of stuff that you can write to get very high performance throughput and finally like I'll give a shout-out to Jo mix there's a really good plugin you might want to use SP DJ mix for for micro benchmarking things in summary I think Scala and ARCA gives you and the reactive stack gives you really awesome abstractions and and performance we used akka you can use aqua for distribution for state for protocols cluster management you can use the right tools like reactive streams for typesafe concurrent stream processing and if you need faster performance than what you know the standard library gives you you can also build very fast abstractions yourself what fairly good API so thank you very much thank I guess I guess we can have a few questions so you said there's a so the question was by minimizing these feel like saying do you have a trade off what record size I would say actually in our case is not true so objects actually use a lot more memory because each each Java object has to have a minimum header size of 16 bytes and you need an additional pointer to point at it which is another four to eight bytes what the binary record basically everything is in line so you don't need to wait to space of additional objects so it depends on exactly what it is you can argue that strings are also more compact because we use say on average of eight bytes versus sixteen byte sorry eight bits instead of sixteen bits each but you can argue that in some cases it might be slower to process a native beauty of a string instead of a sketch other string which is very highly optimized but I think the trade also probably worth it of I mean I think basically that utf-8 strings that are encoded are basically and compatible with essentially like C strings you know like so so let's say it gives you more compatible with other languages like it depends what you mean by other languages right like ultimately everything you know that we're doing runs on the JVM you know but if you mean being able to read today as from other languages I think utf-8 encoding is actually better for compatibility it's just not as good in terms of like the special Turin class we have it's not going to give you every method that a standard string class does so it's less compatible from a Java point of view all right I'm available for questions [Applause]

scala.bythebay.io: Evan Chan, Building a High-Performance Database with Scala

Recording: scala.bythebay.io: Evan Chan, Building a High-Performance Database with Scala

hey yeah so um welcome to my talk my name is Evan and I I've been so thanks very much for the last speakers by the way that was a great introduction to lead up to this talk I'm gonna be talking to you about building a Scala database a high performance database using Scala akka and SPARC just a little bit about myself I've been a longtime user and contributor to spark and Cassandra since the O X days how many of you guys remember the really early days of SPARC anybody anybody know a few people also yeah it's amazing how these communities have grown like from like nobody had heard of it to something that like pretty much everybody is using so pretty pretty amazing I'm the creator and maintainer of a couple open source projects the spark drop server which I won't talk about today but it's a restful api for managing your spark jobs as well as follow DB that we'll talk about today right so yeah they're so so nowadays streaming is king right like so in the early evolution of Big Data what people ran a lot of bad jobs that was really all that MapReduce could could give you and what and what happens is that more and more you find that businesses want insights faster and faster right so so just running like say you know a daily batch job is not it's not good enough people want to know what's happening now right so typically you would have an architecture like this where I'm feeding data events such as the IOT events that previous speaker talked about they get fed into a message queue right and and then from message queue it goes into some kind of processing system where you're maybe doing that you know aggregations or you know that kind of stuff and and after that you need to kind of persist this somewhere because you need to serve that data right so and in the traditionally folks wrote this - - maybe like files on disk or distributed file system like HDFS but more and more that that that is not a really good match for the streaming paradigm as you have to often write data you know different times they will late events you know that kind of stuff so you're starting to see the the riot you know the increasing use now of distributed databases and a couple of notes about this kind of pattern one is that you that you need to always be a pending data but one thing to think about is that it's actually quite emitted teachers to have a property called idempotency and what this means is that it's just like you're writing data to my sequel if I'm writing data using a primary key PK one and some time later I have to write that data again using the same key you notice that it does not get read twice right that it it's it's the data gets written to one place and that is actually that's actually a really good property can anyone think of any reason why that might be good mmm that's right so so basically handling failure right like in a lot of businesses having exactly once ingestion is fairly important so one easy way to think about exactly once it is at least once plus idempotency so that that gets you to exactly once in a fairly simple way you don't have to cook up super complex logic and your streaming pipeline to do that right so a really quick introduction by ODB is basically an open source distributed version call them their database that allows you to do updates and it's been designed for the modern streaming world and let me figure out how to use this correctly so what it is is that it it gives you a columnar encoding and what I mean by that is that it is like a punch of parquet probably everyone most folks depicted water familiar with that technology and you're taking say a million records and you're encoding each column in your schema separately that allows you to apply a lot of advanced compression techniques that work much better but at the same time unlike parquet so it has a the speed like parquet but allows you to filter your data and much more advanced and natural way and as well as ingest data using my primary key scheme like you would in a traditional database and you can and it is basically integrated with spark sequel so that you could query using your traditional sequel which makes it very easy to integrate with standard bi tools like I don't there's like a million of them out there and the way that you would use it is that it's so basically it is it sits on top of Cassandra now partially Cassandra is an excellent database for if you want to write lots of small data randomly and you want to read it back randomly an example as Ford dashboards right so you have a lot of small things that you want to feed to a lot of users and it's very low latency it handles that case really well we're concerned a does not work as well as if you want to use it for things like doing ad hoc queries that might take a long time or building machine learning models any kind of analytics applications where you need to read a lot of data Cassandra is optimized for a use case and that's where it probably comes in is that you can use it as a store for reading a lot of data at once and it is 100% reactive so today's talk is actually more about how we're going to use how we use our favorite technology Scala akka and SPARC and I'm proud to say that that hundred percent includes even the high performance code I think you'll find very very very few Scala data you know hundreds and Scala databases out there and this might be only one I think I'm aware of maybe one other project I'm not really sure so why would you use Scala akka and spark right so I think probably everyone does everyone heard of akka pretty much right so the actor system it's a great model for for a lot of things one is that we can eliminate a shirt mutable state you know which is a easy way to shoot yourself in the foot so there's a lot of state in a database when we ingest data there's stay you want to you know you want to cache stuff so that so that makes a connector fit and alcohol also has a great remote and clustering module that allows you to coordinate multiple notes again things that you have to worry about in a distributed database and in mix building client-server things very easy you just send messages from the client to the server and you can respond there that kind of stuff building back pressure handling failure these are things that again are fairly easy to do with akka and for spark what spark gives us is that it integrates with so many different data sources so it becomes very easy to ingest data it has a sequel parser you know distributed data core processing so basically SPARC handles all the data processing and query needs I'm going to go really quickly into architecture of one what one philippi looks like what it looks like I said at the very top we have a bunch of actors where both data and commands are coming in and then the work gets farmed out to more actors and there's some there's logic underneath that then takes data that comes in as Records and converts it to efficient columnar chunks and one one question is that I'll cover a little bit more depth it's when you use akka actors or akka and general versus using something like Scala futures so the way we do it is that the core the core logic is that is not contained in our actors it autological handling IO and stuff is relatively stateless and uses type futures and that gives us a great easy way to compose that logic you know go from one state to another to produce results and it's completely synchronous which we want for high-performance i/o and what we use akka for is basically two things once managing state and another one is for handling the remote connections like having clients write data in to akka and so there's for example each of these guys has a state like mm table and it's and the scheduling the scheduling is another thing that akka is really really good for actually basically I had to summarize this slide yeah so basically for akka extinct messaging handling state making sure the state doesn't leak right scheduling things it's really really good at what might we might want to for example flush data after a certain period of time so I can support the in scheduler that's really useful for that so this is what it looks like when we send a control flow message and this is a bit unconventional but it's pretty cool so every spark application has what is called a driver or an application and and it also has workers and the sparkling go it's called executors so you have let's say you have a to note a cluster here executors what we do is in probably be is that we actually embed the logic within spark and it spins up in a cluster which is in parallel to the spark cluster and we take advantage of something called cluster of singleton this means that in your whole cluster I want exactly one copy of an actor and that's because this might have state that contains how the data is distributed amongst different notes right and and so when I want to send a command so just flush right from the driver what we do is that we go through something called a cluster a singleton proxy that knows how to find this cluster actor this I mean I have that drawn in middle but it's actually sitting on one of these notes and so it will find the right note send it there and then this will distribute the message to to the different notes wait a minute I think it went okay yeah that's right so so why do we use Atkins spark basically there are a few problems that spark is great for stateless processing like you take data in HDFS or kisara or file ODB and you're gonna process it through a bunch of steps but if you have to keep state that's not something that spark is really good at and for us we have a we have an additional requirement that the state has to be sticky meaning that we want to stay to persist on the same note so if I send keys K 1 and K 2 and K 1 is processed on node 1 and K 2 is processed on you know kate no - we want subsequent keys that are the same to go to the same node so for that kind of stuff acha is a better fit because it manages state and also gives us the asynchronous messaging pattern whereas a spark is not a messaging platform right like you can't you know I there are some hacks I can do like I can send data to audit notes but it's the spark is not really designed for messaging right so so we're basically using our code or complement spark and give us more flexibility in terms of that doing complex things like managing state and the way that we take it we also take advantage of this for routing data so the data comes in at the top goes into a row source actor that then uses a partition map to route data to the right node and process it and then one really nice thing about using actors for this is that in the future let's say that we want to change the architecture and let's say that I want ingestion logic to be separated to a separate JVM now I don't actually need to we architect my app at all I just basically changed our configuration and change where the actors start it and boom they get sent over the network I mean there are some things you would want to tune in like the Java serialization is very slow you probably want to tune that so that you use something faster but but fundamentally it's a very powerful model for building your applications that you can abstract them and make them distribute it we built this before akka streams was really mature so we kind of have to bake in our own protocol for doing at least once and back pressure and the way we do it is that we kind of use ax that's a way of slowing down the data so that AK is really common but then we also built in a retry mechanism into the protocol so that if we don't get ax back within a certain time out then we resend the data and again what a basics that akka offers you such as scheduling helps helps a lot in this area alka has a really awesome testing framework called the multi node spec and an SPT plugin called SPT multi JVM it this is a really awesome tool for building and testing distributed applications it has barriers so that water nodes can reach a certain point and then it will go and you can run tests you can test inject network failures so there's a lot of really powerful stuff in there I'm gonna skip over this a little bit basically the core API is has futures in it so one thing that a database needs is monitoring and performance measurement right so what what we use is a library called came on how many people have heard of came on awesome so k1 gives you are both statistics that are specific for things like akka and play and so on out of the box but it also gives you concurrent tracing which is super cool and what I mean by concurrent tracing is in this code I have basically an append segment this is one or one of the core paths core right paths and you can see in this for comprehension that there's a couple stages and each of the stages of we turn a future so these are composed futures we do one thing and you know in a future and then we you know do another thing right and what you see here is that I'm actually passing a trace in context to each step and what k1 will do is that then later I can sample the traces I can say sample 1% or traces I can actually get even though futures are crossing threats I can still get statistics as well as individual traces like okay this is helping a time a then it went to this out of thread and a time B and this is invaluable for for being able to capture what's going on in an asynchronous high performance system so so this is a really awesome library and one other note is that they're also using HDR histogram which is pretty awesome for capturing histograms and just a really quick note about validation I'm going to plug build vendors galactic library that lets you compose validation things basically it gives you a return type of like you know some type a or some error type and you can compose this all the way down so that at the end I yield the desired result but if it fails at any one of the stages I can easily get out the error type which is which is pretty cool okay thanks finally there's a couple of things so we're talking about high level how things work I'm going to bring you all the way down to you like you know very low level stuff so there's some some interesting stuff that I'll point out like you can look it up and the slides we want but basically what the format that the calmer format we use is also our hand Road basically what this is you can think of it as a binary format sequence or a scalloped vector that gives you random access what note is the realization so you can basically read stuff at machine speed half a nanosecond per read this extremely hype of like public highest performance scholar code that you'll find anywhere and it can work you know its design work like off heap and support missing values and so that that's this is some of the stuff that we used under the covers to get very high performance and we also have a a a generic record type that is also very high performance and it's again a noisy relaxation for on/off heap and all the stuff you know just designed from the ground up to minimize GC and aim for very high performance so who said Scala cannot be the fastest thing on the JVM right so yeah thank you very much it's thank you for the Scala community [Applause]