Scale By The Bay 2018: Evan Chan, FiloDB: Real-time, In-Memory Time Series at Massive SMACK Scale
you hey welcome everyone we're going to be talking about time series today and my name is evan i feel free to follow me on twitter you know stuff photo random stuff especially if you like photography so I've been doing big data for a long time and like Scala for something like five six years and done everything in a big data stack pretty much but being especially involved in SPARC and Cassandra communities and and recently I've been working on time series at scale so that's one won't talk to you about today about developing and and scaling out a time series solution and what we learned so so time series what are we really talking about you know the world is filled with stuff like this who cares about operational metrics in the room it's a show of hands that's pretty much you know most people probably so we all want to see what's going on in our systems right what are some requirements that we might have for for a solution that deals with a lot of metrics and time series the first is might might be scale right it is is becoming more and more common where you have you know you're tons of connected devices in the world you have even small even small companies has could use they have many many thousands of you know machines or more in the cloud you know anybody can scale up easily using cloud services now right so so for the first thing that we're talking about it is that we have a massive amount of data that's coming away that we want to analyze very quickly resiliency and uptime are very important things for these kind of systems especially in the area of operational metrics and folks are worried about their systems going down then your system has to be the one that is up all the time right that's that's an important requirement real-time this word is thrown around a lot what do we really mean by real-time what I would consider real timers is it's basically the time that it takes from when an event happens to when you are able to observe it or know about it right and for things like operational metrics you really want this to be in the seconds as low as possible right because as soon as something happens you know people you want to be able to alert on it you want to know about it and understand see to me latency is more about how fast you can get your answers and you know so real time is that it shows up quickly Liv you know latency is means love different things but in this case what I'm talking about is that you can get at the answer you want from such a system in a very low amount of time imagine you're thinking of like these systems usually being used in say dashboards so you might have thousands of dashboards right and you might have many doubters of alerts how quickly do you think queries need to be to you know make a dashboard display reasonably effective what would what would you guess I mean usually you're talking about something that you know usually it's like say and you know no quite substantially so ii like i think you know you would typically like 199 percentile to be no worse than say you know a second which means that typically you want things to come back no substantially so ii and the other thing that is really important it's becoming more and more important maybe in the old days like you know you would say send 10 metrics and maybe you get back that one met you know the one metric you're interested in or the you know two or five whatever you plot it but it's it's becoming the world is becoming increasingly more complex where folks are looking for more complex answers they're looking for correlations they're looking for more complex queries so so now these systems like time series systems are becoming more and more ad hoc and some feedback that we got from users of existing systems is that folks want a flexible data model so they want to be able to tag their metrics with different values so they want to be able to organize their their data their metrics in different ways and to be able to to query across them and do not be restricted by certain data models like for example a higher code data model would restrict you to currying at certain levels of the hierarchy but what that's not what you just wanted the other feedback that we got from from people was that folks wanted longer views of granular data an example of this is that you know it's Monday morning you come in some some something happened you know and you kind of want to look back maybe it happened on say Saturday morning so you want to kind of look back and see what happened if you're getting something that is down sampled then it becomes harder it might become harder to debug something because you're looking for certain like you know thing that's going on so people wanted to see fine-grained data for as long as possible you also wanted to design for the cloud what does this mean so that the world is moving towards things like Hoover Nettie's the world's moving towards stateless containers and and this has two major implications one is that if if you want to run your thing on the cloud increasingly you're going to leverage hosted storage solutions so you're looking at things like dynamo you're looking at like more and more people are letting somebody else you know manage your state right so that so that's that's one design criteria weren't designed for the other thing is that because containers are stateless systems you have to be prepared for them to be started sorry we started know at random and this actually creates a lot more load for a monitoring system increasingly like what how do you identify each metric it might be you know you definitely identified by application but increasingly like the container ID might be something that's important and every time something gets restarted you have a new container which leads to more and more metrics ultimately you see that where we're going is that we have we have things flowing into the system events metrics you know tracing IOT and we're trying to get stuff out as quickly as we can and as real-time as possible so what goes into this cloud and you know how can we build it ours are some themes of today so I'm going to I have the pleasure of I say reintroducing you to follow duty because some of you might be here from last year I'm not really sure I talked about our project file A to B but we've taken this in a different direction it is now a disputed in-memory time series database which is geared for a super real-time very fast insights a kind of stuff for the massive number of entities which means huge amount of metrics and it is open source so you can find it there it is built on a couple of core principles one of them is that it is designed for the cloud as I'd mentioned that designed to deploy into cloud environments as well as as to monitor cloud infrastructure built for scale that I don't really need to explain this too much we'll go into this in a bit more detail but it has two horizontal a scale and be resilient to failures two important points though are that we want systems to be multi-tenant and to have flexible data model so this is because if you want a system to work in a large organization then you can't really take a one-size-fits-all approach some folks might have legacy data that are of a certain format some folks might have different kinds of methylase and so you need a system that is able to adapt concurrently to these different requirements and this system the time series database is proudly built on the stack that we talked about this conference Scala akka Cassandra Kafka and those are some type safe sorry type level symbols if anyone knows that that this one is monix by the way I don't know if anyone recognized that but pretty good stuff so let's talk about solving the time series problem what is the time series problem this is a slide from prom con so think of it like you have tons of let's say like million different time series that are coming in on a regular basis what is the right pattern like how is data coming in so you'll find that it comes in kind of like it comes in vertically here so each of these are a different time series and you might get like one point you know every 10 seconds or something so it's kind of coming like this right it's going to do the data then it goes did it I mean not like in sequence like that but but overall it kind of the right partners like this the problem though is that what you want to read this is not this is not the pattern that you want to read right then in fact the read parents are almost exactly opposite you find that I you know I might be looking at you know one series of data then another series of data and kind of doing things with it so read and write patterns are almost completely opposite and when the number of items that you're dealing with aren't many millions this becomes a problem like what are you optimized for if you optimize for writes then you might end up what's somewhere is very inefficient to read in in the same way right if you optimize for reads then it might become very expensive for rights and time series databases or after all if they have to do anything it is that they have to scale insanely for for an insane amount rights so so this is a fundamental problem of TS tvs about three years ago Facebook came up with a landmark paper called the gorilla paper and they took a new approach to solving this problem what they said was you know most people are really interested in recent data so what we're gonna do at Facebook is that we're going to keep this data in memory and we're going to use very efficient and coding techniques to fit as much stuff in memory as possible so that I can basically answer a lot of complicated queries very quickly using using this in memory data and we serve queries using a separate process so this allows you to fit a lot of stuff and answer queries quickly and solve this right powder and versus read pattern problem they open sourced a look at like a tiny percentage of it unfortunately so basically all of us kind of ran around and scrambled to see you know how to build it ourselves you know around the same time the Prometheus project came about how many folks have heard of Prometheus so that's interesting it's not not everyone in the room so so for those of you I guess I should give a five-second so Prometheus is is a very popular time series solution that I'll talk about in a little bit it it came out around the same time using very similar principles they were going to store whole bunch of recent stuff in memory and use efficient techniques it is however it belts for it is built for the Google Bork Mon kind of architecture which means that every team is designed to run their own Prometheus's so it's designed for it's designed to be a single node by design it is not designed to scale which has a lot of implications and this didn't really work for us so we decided to build our own do build is something that is scalable from the ground up and this is what a dataflow might look like so imagine that you have in one scenario applications that are sending metrics operational data to something that we call a gateway the Gateway instance then charts this data into a queue Kafka Kafka's role is as in many systems used for charting buffering as well as we play of data so that you don't lose data one thing that I'll note is that it is necessary to accommodate different ways of ingesting data so you have push going on and you have pool going on for different people different kinds of network infrastructure means that you know one or the other is going to work the data is pulled from Kafka into a set of independent Philo to be notes and the data is then served rapidly to various users ad hoc users dashboards alerts now let's look a little bit at how this data flows in so data that it's ingested at a high volume is a pretty good fit for streams right so we actually use monix which is an implementation of the reactive streams portico the you've probably heard that many many times in this conference we like monix a lot because it gives you the standard reactor streams and back pressure data comes in it flows it gets indexed it goes into what we call right buffers which is a key abstraction for solving the right andreat pattern difference is that data gets collected and then encoded into efficient chunks we prime series and it gets written out if there's any problems in any of these feeders what the reactor streams paradigm means is that something that slows down later will also slow down to read and and that means that you can slow down the rest of the intake so that you know again you're not like losing data and one thing that monix is really good at doing is that it gives you this complete typesafe backpressure api that it's very easy to manipulate but at the same time monix is really designed for high performance so you can control the threading and a lot of details very precisely which is a big bonus so so late it comes in you might ask how is it that we managed to fit a lot of data memory and we do this using columnar compression so I trait so a traditional record way of storing the data would be that let's take a really simple schema where for each point that comes in it consists of a timestamp and a value let's say this valley was a double for purposes so a normal way of storing this maybe I have a case class right which is points a data point and that might have a timestamp and a value so they would be stored next to each other in memory and you would have a whole bunch of these which is which is fine this is the most law a way of representing this data the only thing is that when you store data in this way it is it's it's it's really hard to compress it so what we do instead is that we store all the timestamps together like t1 38 here and the values together v1 through VA when you store data of the same type together it's much easier to take advantage of techniques and for example we can look at the time data and we can use oops yeah that's right so you take the time data and it you would you might say you know what time stamps for time series are likely to be increasing smoothly so let me plot it on the graph so to speak and do a linear regression on it and I will fit it to a line and now I can represent this data as an intersection and a slope and the Delta from R to the slope so this technique is called Delta Delta encoding because we're taking the deltas and there's also a delta this way and what we find is that because of the mostly linear increasing nature of timestamps that most of the time you can encode the deltas which are significantly smaller you can take this a step further and say you know maybe people don't really care about timestamps below a certain resolution like you know a quarter-second right so we can kind of throw away those deltas that or below a quarter-second and so you can do a lot of things like this to significantly reduce the amount of space that that you use and the result of these techniques of storing data and memory using columnar compression Delta encoding and similar techniques and a lot of optimization is that we're able to store millions of time series and many many billions of data points in memory on a single node we can actually ingest up to a single node can ingest up to a million data points a second at peak which is I would say pretty impressive for a Scala JVM language like this is not written in C++ books you know so and this was this is about an order of magnitude better than previous system which used storm and HBase and we have we can achieve storage ten cities which are fairly good for amount of data would say around say you know three bytes this varies but I mean this varies a lot really depending on exactly what about the data it's like but no like a couple of whites per metric sample which is a lot better compressed and raw data and because you're storing everything as a column it lets you use a lot less space quick question so so Corolla well goal is not something that someone can use I would say we're not quite at gorilla scale yet you know we have about I don't know 50 percent or something to go but you know I think that's something we can improve over time so yep so you might ask just doing so many things how do you store all the stuff right so tackling heap issues it was is that was a major area of development for this project I think at the very beginning you know we're using so much memory when you build things naively that we ran into like really long pauses and what that is is that because you're storing a lot of data and if you store it on a heap it gets collected in in your old generation and it's most most of the Java GCS are really built for quickly cleaning up recent data it's because you can go through young gently the old gen it has more objects and it takes longer to go through so so that takes longer and you can get into loops where the got stuck because you know you are you have so little free space that it's not able to work and then you get into really long pauses so so our solution was to try to move as much of this data off heap as possible which eliminated pressure on on the old generation the other major problem that we ran into is that the rate of ingests was too high and you know maybe you might run into this in various streaming systems as well where you're allocating things a lot on address tchen you might be deserializing you might be doing things what we found was that we use protobuf initially for the ingestion this ended up actually taking too much space because you can imagine for each probe of object then might lead to a graph right so one object might contain like say a map you know and and that might be a list of tuples each tuple might then create an object so this kind of balloons very quickly so we created a custom data structure that allows us to do zero allocations on one ingest there's a couple of these that if you're interested you can look for in the open source code one is are compressed columnar format called a binary vector which is kind of like a vector in scala it's it's a rentable list and then but a Nick which can be compressed and read out in a compressed form there is a record type which can support different schemas and allows you to ingest data quickly with without allocation and they use something that we call mem factory where you pass this in and just passing in different mem factories allows you to specify whether you want this data structure on heap or off heap and exactly what kind of allocation strategy that you want to pursue so this gives us a lot of flexibility mooching moving object graphs off heap so is this a quick example of some of the optimizations that we did initially we used a concurrent skip this map which is a wonderful data structure well it is a wonderful data structure until you need to have millions of copies around so a a concurrence cubist map has an entry or a key value pair right so the key has to be an object and the value has to be an object too and for us the value pointed at other things which pointed at chunks so this created an object graph and what we found was that even after moving certain things like these these chunks to off heap if we the normal pattern of doing this in the JVM is that you would have you know classes and instances for referring to these things but each java objects takes up at least 24 bytes of memory and if you multiply this graph a times 24 times millions and millions of objects it still ends up using a huge amount of space in fact the amount of object overhead that was needed to store all this became a thing of caen fraction of the actual data that we want to store so what we did is that we moved this stuff into a bunch of a few data structures that points at different things and this allowed us to eliminate the overhead as well as eliminate the GC pressure so a system like this is not really useful unless you could query it right we might be asking how do we create something in a way that it allows for a huge amount of concurrency so what we did was we took a departure we used to use spark as the query layer but for this project we decided to build in Prometheus query incompatibility this is nice because folks that are in the monitoring world like SR yeast they are they are used to the Prometheus query language if you're not familiar with it I can give you a really quick overview but supporting a popular query language is a seems like a good thing because we don't have to we've met a lot of other things and we can get out-of-the-box support from popular projects like Cortana so this query suppose that you want to find out for your HTTP service you will find out you want to find out for some given data center how many of them have occurred and you want to find out maybe the top hosts that are giving you those requests you know or you wanted to scoop them up a house so so here this is like a really quick prompt ql query what it does is you're putting a metric name in here actually requests and an inside curly brackets are a bunch of filters so the filters you can think of them as key value pairs that filter out the time series you're interested in so in this case I'm looking up a particular data center I'm looking up a particular job number and I'm going to sum them I'm gonna do group I by by the host and then I'm going to sum them up so so this is a really quick or some synced syntax that allows me to do a whole bunch of standard time series aggregation so the whole thing is built on time windowing which is another really important thing to you for a TS d D what you do is that once you have a query first you have departs it into a logical plan so the logical plan at the bottom there might be a bunch of filters that one of them is this thing called underscore underscore name which is for the metric name and it gives you audio a tree requests then you might filter it some more by data center and by job and things like that and then we might do some kind of time windowing on it and we might then sum it up at a certain level now the logical plan only tells us what to do it doesn't really tell us where to carry out this work or exactly how to do it so logical plans are then translated into a physical plan now for us our system underneath is akka based it is based on actors and this is an important part so the way that our fiscal plan works is that each step is carried out at a certain actor which is mapped to a shard for us so in this case the fiscal plan consists of carrying out a lower level selection of time series and aggregations at each chart so this happens in a distributed manner a higher level of the tree will then take two intermediate results and sum them up now one really important aspect of of this as related to reactive systems is that we take four vanish of aqus location transparency how many folks know what location transparency is let's say like not as many hands as I thought right so what so the way that akka works is that it's an actor motto meaning that everything is an actor to which from which you send messages all you need to know when you send a message is the actor reference of another actor it doesn't matter if it's local or remote and that's what is meant by location currency is that the location of the destination actor doesn't matter it can and that is they really key concepts in our physical plans we would have an active reference and so you can usually change the layout off your query execution simply by assigning a different actor and changing changing and you know to match your the desired topology for example maybe it's okay to do this upper level stuff in short zero but maybe we decide tomorrow that you know this is not really optimal it leads to unevenness in CPU or whatever and I want to do this in an external service so I'm going to move out this code pure query service well if I'm running things using actors this becomes completely transparent with no change in code because all I have to do is I want to say I want to carry out this step in in the actor over here and and that will let me carry out this plan using this new topology and I can test it out really simply in fact you can even dynamically decide what is the right typology based on a weighing of the cost of executing our query which is which is really nice our actor Heike looks kind of like this in each node you know at the top you might have say an HTTP running naka xep and we would have a note coordinator which receives auto messages the no coordinator will then forward things like say a query to underlying level of actors so there would be a query actor and this would then parse the krei it would look at execution plan and say oh you know I can carry I'm supposed to carry out this child plan on this other note so then it will just forward that request to another node and that gets carried out so this is how we can have dynamic query execution that that we can optimize at runtime I'm going to look let's look at the data model a little bit because this is sort of important to understand because a lot of people talk about time series and the term is thrown around a lot so what what is this system like this optimized for it would say it is when you have a very high cardinality of time series and exactly how we model that is up to you but for us we would model each individual metric from a single instance of an application for example would be a separate time series and they would have points in time so so basically you can think of the time series being the most important part of your primary key so to speak and time being the other part of your primary key and a lot of things are keyed off of the specific time series and the kind of tax that you might have that the indexing system is based on this so this is how you look up things and filter things so so this this becomes so having as many series as you can for fine grained data is is is a key part of it and this is sort of opposite from a traditional like sequel database or whatever were you might have like a few tables you know maybe have five maybe you have ten but you're not gonna have like a million tables right where is this kind of and what when I say time series database the reason why this is very different from most almost any other kind of storage or database is that it gives you the ability to track individual entities like a very fine-grained level so I don't have to say you know let me group like these hundred thousand you know phones or whatever you know to do this table and then fit this other thing to this table I can just track like each and every one and not really worry about charting and things like that the tags are done in such a way that any unique combination of keys and values automatically forms a unique time series so let's say that I'm sending this time series that has short equals zero then I decided sled one where I have data for short equals one the net on math becomes a new time series we have a name spacing mechanism called data sets so probably B is different in that it's a multi schema database that we support different kinds of ingestion schemas and this is used for the dirt this is an example of how we might namespace things for and this is a big help for multi-tenancy so suppose that I have a main data set which has a two day-to-day retention policy I might have a different one for what we call pre aggregates these are data points that are pre aggregated in a certain way to speed up craze and because that uses up a lot less space we might decide to store that for a lot longer we might have one that especially for data with big histograms complex metrics like such as histograms end up being a major consumer of space and i/o and storage and things like that and maybe I have one you for historical data that uses a different schema and so this gives the flexibility to support a lot of different use cases at the same time now the really interesting stuff for recovery and persistence what kind of stuff is persisted well the raw time series data is persisted using Kafka and that's the the raw metrics that are coming in and the compressed Colin our data is persisted currently to Cassandra and then we also persist some indexing data so let's take a different view of things and look at it from a clustering standpoint we use akka cluster connect different notes our cluster is used for shard management and dynamic reassignment of nodes and failure handling and things like that so what we do is that we map the data for each shard to it to a Kafka partition and then notes are signed shards we actually don't use Kafka's automatic dynamic assignment because we have so much state in each of our notes that we don't want things moving around with high frequency as you as might happen what the default Catholic liar so this is the weight of things get then we assign things and then we start pulling them into notes so what happens if one of these fails right so let's say that this node that has sure to fails then what happens is that let's say you have a new node that comes up the clustering would recognize this we have a wave a singleton this would assign this node that data it would start recovering data from Kafka we actually don't read the data from Cassandra automatically this is read on demand and the reason for this is because time series data is usually write heavy and the reading is a lot lighter so we just read it from Cassandra on demand which matches the Cassandra data model well because it is a key key value and what what do we learn from audits right think about like it's pretty rare I guess for folks to be doing this is using the JVM so I would say you know pluses or that we have the huge ecosystem of JVM libraries that we can rely on our cluster Lusine and very reliable clients for or major dependencies things like that and the the minuses are that you don't really have a a really good fine-grained control of low-level things right we get it done today it's just not anywhere near as pretty as it could be there's a couple of things that would be really nice over JVM wish lists for for things that would help out a lot such as having value classes there are some turn things in Scala that do help such as Scala does have a value class but it's not a true class so there are certain things like that that would help having actually safe typed pointers that that would be a help that would be a big help that there are certain other things like utf-8 native utf-8 strings that are coming to future versions of JDK that what also helped some of it is in growl that has been talked about several times in this talk a lot of it is in project Panama and some of it I'm not sure where that's coming and just to go over this really quickly we have it's really good to get to know your GCS really really well and I was just point out one tool that we like a lot which is called sjk this is call it stands with swiss java night and it lets you gather all sorts of stuff from heaps to cpu analysis to you no threat allocations for our production systems today we are our development actually happens open source a lot of people don't notice but Apple sorry we the development actually does happen at github and we're looking for partners to partner with us to work together and so I encourage you to try it out you can actually gather metrics and look at it using a graph on our dashboard and lastly I'll just point out one thing I wrote on the world map one thing that is of great interests is how do you get better observability for your data pipelines like how can we help data engineers a lot of these tools oftentimes like the data engineering world and the scr ye and monitoring world are kind of completely separate you know and most data engineers have to figure it out themselves but how can we have a tool that can actually help analyze your bad job performance your streaming job performance your models you know your you know stuff like that what is important in that area that that is one thing that would love to have feedback on so looking forward to talking to you thank you very much I'm sure there are questions right so question is what what kind of data oh that yeah I can probably like it's probably not a quick quick answer so I can talk to you about but for now for the open source version we support a and a certain ingestion format that is very popular in the metrics world yeah it's not very complicated we simply look at you know we have sort of a resource manager kind of thing that we just look at here is available notes and you know how we can assign a science stuff you want to achieve concurrency yeah yeah you're absolutely right I'm not sure I have like any like magic tricks you know there are people I think that I do think that tracing is pretty important for our disability systems that you kind of see where data goes and you know when things happen that that's one thing I think we should keep improving support for fortunately that they are there are there are a lot of libraries that are really helpful in the ecosystem like we use came on for example and came on as an open source debugging or monitoring system that actually covers acha and you know some other things like that so to help you monitor your actor mailbox and things like that so we just take advantage of as many of those tools as as possible and add a lot of visibility ourselves yeah thanks Oh some more cool some more questions yeah so your question is are we making queries directly on compressed data yeah so actually today our queries run directly on the compressed data the the current format is done in such a way that we don't have to decompress to to read from it so that's kind of an advantage I essentially created a new time series how does that fit back in sort of the query language and filter aggregation so like this query dong right the main theme of the of the set of series but it actually included 15 you know subtags right yeah those would all be put in so for example if I if I create on dataset equals time series they would pull in both of these series and this could it'll find all them and right exactly exactly yeah yeah yeah that's one of the things that is powerful while this approach yeah okay which would cleave one more I don't know yeah okay yeah I mean it's it's kind of difficult to like we do have those numbers that are shared earlier like when it comes to like the depends on what aspect you're talking about I mean I think the numbers here are that I've shared are comparable to Prometheus in terms of like ingestion for example but it's really hard to compare because for example meat is single it's is a single note and and art queries are distributed by default so it's sort of difficult to end them all and the day tomorrow is similar but not exactly the same so it's not like easy to you know compare you know I'm available for your questions if you know outsider [Applause]