Bay Area AI: Succinct: Enabling Queries on Compressed Data w/ Anurag Khandelwal
Recording: Bay Area AI: Succinct: Enabling Queries on Compressed Data w/ Anurag Khandelwal
[Music] hello everyone and uh thank you Alexi and Rael and Adam for having me here and I'm really glad I can talk to you guys today and very quick introduction I'm anurag I'm currently a PhD candidate at UC Berkeley and today I'm going to be talking about succinct which is a A system that we've been working on at amplab to enable fast interactive queries at scale on huge volumes of data and this is a joint work with rajit who is currently a uh professor at Cornell and my adviser Yan sta at uh UC Berkeley you might also know him as the chairman of data bricks and the company that started uh spark and I'll just Dive Right In uh I don't think I have to convince given the the audience I have here today I don't think I have to convince anyone of you that we all want to uh perform complex analyses on uh larger and larger volumes of data and we want to do so without giving up on performance and in fact interactive latencies often correspond to uh latency of the order of seconds or even subsecond latencies and uh no matter how complex your analyses are they can always be decomposed into different kinds of primitive operations as such including operations like search you might want to uh look at all the tweets that have been made by amp lab about succinct uh regular Expressions where you might want to you know filter out all the documents that contain links pointing to domains from uh Stanford or uh berkeley.edu range operations you might want to actually uh filter out tweets or Facebook posts that you've made between 2013 and 2016 or you might want to perform complex operations and graphs for example in a social network you might want to look at all the friends of your friends who enjoy trekking and analyze that data in addition there are a bunch of uh seemingly simple operations like aggregate queries or random accesses and updates which become challenging given the scale of the data that you're performing this on and to cater to all of these different kinds of operations we've come up with uh a variety of increasingly complex systems for instance we've built compute platforms like spark and hop which enable uh arbitrary computation on huge volumes of data we've also built query engines on top of these compute platforms such as spark SQL um Cloud era and uh Cloud era's Impala and hbas which allow us to execute SQL and relational operators on these Compu engines and finally we've come up with data stores uh at one hand you have data stores like mongodb and Cassandra which enable really low latency and high through put read and write operations uh and on the other hand you have search based uh data stores like elastic search and solar which focus on search based operations and more complex operations like regular Expressions while maintaining interactivity now uh most of these systems are pretty good at what they do I'm going to say most not all because we always have some systems that we don't like but uh the problem the uh I'm going to in this talk I'm going to focus on two problems that almost all of these systems run into at some point or the other the first problem that I'm going to talk about has to do with degradation of performance when the total amount of data grows much larger than the am amount of memory that you have in your system to show my point we looked at a generic system uh uh that caters to Big Data and uh the name of the system is not important for this demonstration purpose but what we did was we measured the performance of the system in terms of throughput that is the number of queries uh that it can serve per unit time as a function of the total amount of data that it has to handle and what we found was that essentially as long as the data can fit in memory your performance is pretty decent as soon as the data grows out of the amount of available memory the performance takes a drastic Spike and you see uh severe degradation in performance next perform the next uh problem that I'm going to talk about is catering to skewed query workloads and the uh setup that we we're going to be working with is a distributed system that scales horizontally using uh partitions so you have multiple partitions residing on multiple servers and the problems that we look at is uh the low distribution across these partitions in the distributed system is non-uniform in fact it can be heavily skewed towards one of the partitions and the problem that arises is uh you have a maximum sustainable throughput or sustainable performance for a particular partition and there's a degradation for these systems when the skew results in the queries going much much overboard the much much over the maximum sustainable trooper and I'm also going to talk about our solutions to these two specific problems uh and those are uh two systems that we built that is succinct and Blowfish which Builds on top of suin to cater to these particular problems and hopefully by the end of this stock I will have you convinced that uh Su synct as a system caters to the first problem by enabling a compress representation that allows a large fraction of queries to be executed in main memory which is the fastest storage and it also enables a rich set of functions to be executed directly on a compressed representation of the data you can perform queries like search regular expressions and uh range operations directly on a compressed representation of the data I'll also talk about how succinct exposes a flexible support for different kinds of data models including uh data models such as key value stores uh columnar stores tables uh and document stores and finally I'm going to talk about Blowfish and how it allows us to handle to cater to the second problem that is handling skew query skew in these uh distributed systems and in fact even cater to situations where the query workloads have SKS that change over time so the the skew across different data items change with time but to begin with I'm going to focus on the first problem that is uh the degradation of performance when the total amount of data stops fitting in memory and as a running example I'm going to use a flat unstructured file as shown in the slide over here which is composed of differently colored blocks and as a toy example the query that I'm trying to perform is to search for all the green blocks in this file the first approach is to use data scans which is prevalent in systems like sparkk or Hadoop where what you essentially do is you look at the data and you scan through each and every block in the data to find all the occurrences of the green blocks now the problem with this approach is the same as with my animation it's terribly slow and at least very high latencies now the good thing about this approach is you do not need to store any additional structures to support scans you can essentially just look at the data and get your queries and therefore it has very low storage overheads the other approach is to use indexes which is common in systems like elastic search or Soler where in addition to your raw data you would store an additional data structure called an index in this particular example the index is an inverted index where you map the differently color blocks to different locations where these blocks occur in the file now if you wanted to look up the uh find all the occurrences of the green block you'd simply look up its index entry and you get the result so the good thing is you have low latency but the problem is you have to store this additional index structure which often leads to very high storage overheads so and this becomes a particularly big problem when you have very already have a very large data set size and constrained amounts of memory so that these indexes bloat your storage footprint even further and lead to your data set not fitting in memory and this becomes a very big problem because you are essentially when you're executing queries is essentially thrashing the cache where the data movement in and out of the cach leads to very severe degradation of performance and quite counterintuitively use of indexes actually leads to worse performance in those situations in order to cater to this problem we've designed snc which is our system uh to take a radically different approach now what snc does is it takes the uh raw input and it transforms it into suit of data structures the suit of data structures forms the compressed representation of the data and you don't need to store anything ex except this compressed representation and what sync enables is for queries to actually execute directly on the compressed representation of the data now what this enables us to achieve is low storage because the compressed representation as well as low latency for two main reasons first of all our query algorithms are actually optimized to perform uh to run directly on the compressed representation leading to latencies and second because of the low storage overhead uh you can always fit much more data in memory as compared to uh the existing schemes which means that you'll have a larger fraction of queries being executed in main memory which means you'll have inmemory latencies which are much lower now what differentiates a sync from the other techniques is that you don't need to store any additional indexes all of the query responses for these complex operations are actually uh embedded within the compressed representation itself second you don't require any data scan so what we're essentially providing is the functionality of indexes without actually having to store them in fact we just keep a compressed representation of the data and finally we don't require a need compression as well for most of the operations in succinct these queries are executed directly on the compressed representation except for a few Conner cases where you actually want to look at the data and you might have to extract uh decompress a few bytes from random locations now if you look back at our initial goals and see how seink fits in seink achieves scale by being able to analyze data data sets that are much larger than the available memory itself it enables a large variety of complex operations such as search range uh range queries random axises regular Expressions Etc directly on the compressed representation and finally it does provides these two properties without giving up on interactivity which means we are able to execute these operations at much very very low latencies uh because we avoid the overheads of data scans and data decompression for almost all queries now I'm going to very briefly talk about uh sing's internal data representation and how actually stores the data to allow queries on compressed data at a very high level suxing Builds on a large body of theory work so we looked at actually looked at a very large body of uh Theory work that has to do with you know uh indexing data structures and compressed uh data structures and the main uh data structure that we picked up was the suffix array and these suffix arrays are nothing but array representations of suffix trees now the good thing good thing about suffix arrays is that they provide very strong functionality and in fact these are full text indexes that support arbitrary compet searches for uh arbitrary patterns anywhere within the text the bad thing about uh the suffix arrays is that they have very high storage overheads and secondly they do not possess any parent structure that would enable us to compress them now the challenge comes uh for that that succinct is presented with is how do you actually achieve compression and the approach that suin takes is to actually sample the suffix array so instead of storing the entire suffix array we store certain samples at periodic intervals and in addition we also store a set of pointers that allows us to compute the unsampled values on the fly now I'd like to focus on the fact that succinct is a lossless compression scheme so you can actually compute all of the uh unsampled values on the flight using these set of pointers and the observation that cinct exploits essentially is that these set of pointers possess a beautiful structure that allow us to compress it pretty aggressively and without going into a lot more of the details this is in a nutshell how succinct supports really rich functionality on a compressed representation of the data now I know I've skipped a lot of gritty gritty details here and I'd be happy to talk about them uh offline after the talk but I'm going to avoid them anyway because of the you know just an interest of time instead I'm going to focus on sein's data model and particular I'll talk about how succinct enables a large number of different data models to be integrated through its unified interface which means that we can uh cater to data with models that could be unstructured or flat files that could be in key value stores that could be in document stores or columnar representations or even tabular representations what suin essentially enables is for us to run the same powerful Primitives that I described that is search regular Expressions range operations on whatever the unit of data is for that data model for instance it could be values documents or columns and to see how we actually achieve this I'm uh going to take a step back and I'm going to take talk about the secret Source behind what enables exing to do so and the secret Source essentially is a single unified interface which combines all the different uh data models and that uh unified interface is a flat unstructured representation of the data and before uh moving on to talk about how we actually integrate different data models to this single unstructured representation of the data I'm going to talk briefly about what kind of operations we actually support for this unstructured data so I have the same running example as before the uh flat file composed of differently colored blocks on the right and on the left we have the succinct compressed representation of the data now all the queries that we support on Su synct actually yield results in terms of the flat and compressed data so for instance if you were to run search it would yield offsets into the uh uncompressed file where the Search terms occur and uh you could perform Random Access into the original data by using the extract functionality which in this example fetches the first five blocks in the original uh unstructured dat uh uncompressed data you could also perform uh count operation which counts the number of occurrences of a particular term and this is interesting because seing performs count at much lower latencies than the search and we found that this enables a lot of neat applications which I'm happy to talk to you about offline and in addition we also support uh addition of new data to the compressed data store using an append interface which allows us to pend new data to the compressed uh representation I'll talk a bit more about the pend operation in particular in a couple of slides but for now uh I'll just quickly mention that range operations and regular Expressions that also supported on the same compressed representation using an interface which is quite similar to search okay okay so uh if I have no questions here I'm going to move on to how we actually unified the the different data models into a single unified interface so uh to show how we actually unify the different data models I'm going to take a very simple example of a Col a table with four columns as you can see here in this example the first step that suing takes is actually assigns a unique delimeter to each of the columns in this data so once we have assigned a unique delimiter to each of the different columns the next step is to essentially transform this structure data that is the table into a flat unstructured representation in this particular example by writing out the different column values separated by the unique delimiters as you can see in the figure the final step is to actually transform the uh flat unstructured representation into suin compressed representation and now you can support all of the queries on this flat unstructured file as I'd shown in the previous slide now you know your uh next obvious question is I had a table to begin with how does it help me to have a flat and structured file where I can run queries that may have no relevance to the original table to cater to this problem succinct actually translates queries on the structured representation that is the table into queries on the flat unstructured representation for instance if you were to search for all the green blocks in colum one in this example you essentially transform this into query on the flat file where you're searching for all the green blocks followed by the column delimeter for column one now if you look at these two queries for a second you'll realize that both of these queries actually yield the same uh set of results and you would actually get the results that you would expect from this query so this is essentially the high Lev technique there sync used for unifying all the different data models using a clever combination of delimiters and flat file representation whether it's a Json representation of data or you have key value pairs or you have column representation okay so that's essentially uh the all about the data models and I'm going to quickly move on to sinks architecture and talk about how sinks architecture is able to take care of supporting updates now yeah how are these so yeah that's a good question so so the question is uh who comes up with these delimiters and what happens when you already have these delimiters somewhere in the data so I I have skipped a few details here you're right to observe that but essentially what we do is we translate whatever set of characters you have in your input into a much uh into a larger and you map it to a larger alphabet so essentially you have some characters that are definitely not present in the input and that allows us to actually use characters that are not present present in the input so essentially we pick these delimeters and assign them to the different data models whatever data model you might have and compress it so that there's no conflict between the data within uh the uh data that you've provided and the delimeters that we've chosen that's actually all right so yeah uh the next problem that that you have to cater to is in a compressed data store one challenge that you have to overcome is supporting updates on the compressed data because compressed data essentially is immutable and the way succinct addresses this problem is essentially using its multistore architecture uh at a very high level suin has three different data stores internally which are optimized for three different operations the First Data store which is the largest store is the sing store which stores the data compressed using suin compression algorithms as I had described now this uh data store is essentially IM mutable and only supports readon operations the next store is called the suffix store which stores a smaller fraction of the data and keeps it in uncompressed suffix representation if you recall I had uh I had uh mentioned that snc builds on suffix arrays as its underlying data technique data data indexing technique and the suffix store essentially just keeps the suffix represent suffix suffix arrays in an uncompressed format and finally you have the log store which stores the smallest fraction of the data which essentially keeps the data completely uncompressed now when data new data comes in it essentially gets appended to the uncompressed representation in the log store and any uh when the amount of data in the log store crosses a certain threshold we essentially transform it into the suffix store we convert the uh uncompressed representation into the suffix array representation and com combine it with the suffix door once the amount of data in the suffix row too large we essentially compress it using seing compression techniques and append it to the seing store and make it an immutable store within the sixing store itself so by using a multi-store architecture we're able to essentially uh translate the updates periodically from the log store to the suffix store and finally to the Su sync Store where all of the compressed data is kept yeah is it possible by policy to say that if the log store gets very large there's some way to truncate it even though this might have some effect on certain queries so the question is uh given that you have some constraints on the size of the log store you should be able to truncate the log store uh yeah so that's essentially possible in snc what you can essentially do is preempt the uh pre-existing threshold of when the lock store may get combined into the suffex store and you can say that at this point I want to actually uh dedicate you know assign some resources to compress the data into the suffix store and therefore into sixing store question was slightly different okay that given the fact it may already be in the suffix store that's no longer issue just if it gets large enough at a certain point for size reasons it might be necessary to truncate so assuming it's already in the suffix store can you then decide by by policy to to trunk definitely yeah I mean uh I probably should have mentioned made this more clear but essentially once the data dat is moved from the log store to the suffix store and or from suffix store to the sing store the data in the original store is deleted so you don't no longer keep redundant copies of the data in the other stores so that's essentially the approach we take to actually get rid of all the redundancy yeah all right yeah question what kind of compression Rao you achon right so uh the way uh I would answer this question is essentially it's dependent on the data set because compression can vary widely depending depending on how much redundancy you have so essentially uh the amount of compression that cinct achieves is a fixed Factor difference from what you would achieve with some uh you know traditional compression scheme like GP for instance we've observed that with different we've looked at a lot of different data sets ranging from uh text documents to Binary data to even uh Json documents and we found that we're almost always within 1.3 to 1.8 uh times of gzip compression so that's essentially gives you a a high level overview of what kind of compression we get okay so um now we've uh now all the techniques that I talked about we have already Incorporated that as a different Standalone system which you know caters to all the problems of fall tolerance and uh data redundancy Etc but what I'm going to talk talk about today is our recent release of succinct as a spark uh as a p package on top of apachi spark and I'm going to assume a certain degree of familiarity over here with spark as a system in that is a uh large scale distributed system batch processing system that uses the concept of resilient distributed data sets or rdds to perform complex analysis on huge volumes of data and I'm going to Dive Right In and say what essentially success syn enables is queries on compressed representation of the rdds so for instance if you already had a pipeline that were that that work with spark what essentially suin allows you to achieve is new functionalities first of all uh we add document store and key Value Store interfaces to uh spark where you can perform complex operations like search and documents or values we also provide falser alternatives for some operations in on rdds for example you can perform Random Access and filters on the spark rdds at um much lower latencies and the main reason we able to get better performance is because we avoid the overhead of scans and that uh spark often runs into and finally uh what applications can straightway achieve is the ability to push much more data in memory because of our compressed representation of the odds and uh most of our queries being executed on the compressor presentation without requiring any decompression so these are the essentially the three things you would uh that sucin enables on top of spark now uh again I'm going to provide a very brief overview of the different interfaces that succinct exposes for spark and I'm going to assume a certain degree of familiarity with how suin uh how sparks uh Scala interface works but if you have any questions at any point you should stop me and ask me the first interface that succinct exposes for unstructured data using uh the succinct rdd interface where it is essentially deals with data that has no parent structure can and uh is can be looked at as a single huge blob and a typical example would be if you have a huge collection of logs that have no apparent structure in them and in order to analyze this data you would essentially start with importing the relevant classes in Scala you would read the data into an rdd as you would for any regular spark rdd and essentially by calling rdd do succinct you would compress it into succinct rdd interface and once you've compressed tested into a succinct rdd interface you can essentially uh execute all the complex queries that I talked about for instance you could run search uh in this particular example you're searching for all the occurrences of Berkeley in the uh entire compressed flat unstructured document and the result is in terms of a collection collection of offsets within the unstructured data where these uh Berkeley tokens occur you could uh count the number of occurrences of the term Berkeley in the entire unstructured file and finally you could extract random uh data from arbitrary locations in the uncompressed representation for example in this particular case we're extracting 100 bytes from offset 50 into the compressed file the next interface that we expose is for the key Value Store representation which uh is encapsulated in the seing KV rdd interface and the first step is essentially the same you import that elant classes for the key value interface you read the data as a collection of key value pairs uh into Sparks as an rdd and uh once you call rdd do suing KV it'll transform it into the compressed uh suing compressed representation of the key value pairs now you can perform operations like search where you can find all the keys for which the values contain the term Berkeley or whatever term you're searching for and you can perform standard key Value Store operations like get put and lead for instance in this particular case I'm fetching the value corresponding to key Zer so you could do the get put and delete operations that you would see typically in a key value store now there are a bunch of other interfaces that we have also exposed including uh an interface for Json documents as well as a data frame interface which works with spark 2.0 to allow queries on uh spark SQL but in the interest of time I'm not going to dive into the details of those interfaces instead I'm going to focus a bit on evaluation and look at how sucin faires against uh Real World Systems and real world scenarios so for the evaluation use used a Wikipedia data set with 40 GB of data we used a distributed cluster of five ec2 machines with 30 GB of RAM each these five machines were slave instances there's an additional Master instance that control these slave instances instances the workload we used was entirely composed of search queries where we searched for terms that had 1 to 10,000 occurrences within the entire carpus of data and the system that we compare against was spark without succinct so spark is a stand loan system and uh we also compared against elastic search and uh this uh implementation of the this Benchmark was not within spark so elastic search was isolated from spark without having an overheads of talking to spark and uh a few caveat that I point out is these are absolute numbers that are data set dependent so I would like you to focus on the relative Trends rather than the absolute numbers which you might which might differ depending on the data set or the cluster setup that you might have so if we look at the search performance uh here on the y axis I have latency in milliseconds in logarithmic scale so if you look at the latency uh spark uh when executing search queries either on disk or in memory uh yields a very high latency and the main reason is because it has to perform full scans of the entire data set for executing any of the search queries elastic search on the other hand can achieve much lower latencies than spark because it uh can use these inverted indexes to speed up uh you know SE search queries and i' uh point out that uh all of these systems can actually fit their entire data set in memory the performance for elastic search would be uh significantly worse if the data did not fit in memory and finally we have succinct which can uh get even better performance than elastic search uh by executing queries directly on compressed data now I would point that the difference between uh elastic search and succinct performance is not fundamental in that these uh the benefits that sixing gets is by avoiding some of the system implement mentation overheads that elastic search has so essentially you could come up with a system that uses inverted indexes and gets a better lat latency than sixin but the key takeaway here is that succinct on Apachi spark is almost 2.5 times faster than elastic surch and this is while we achieve a storage footprint that is 2.5 times lower than elastic surge again these numbers the data set dependent you might get higher or lower gains depending on what data set you choose to use and again this is when the data set fits in memory for all of the systems another feature that we've recently added and I'm really excited to talk about is support for regular Expressions directly on the compressed representation of data in compressed rdds so the motivation is quite clear you have a large number of applications ranging from data cleaning and information extraction all the way to uh bioinformatics and uh even document stores that use these regular expression queries and what essentially we support is all of the different regular expression operators that you you might see like Union concatenation wild card and repetition operators directly on the compressed representation within the rdds and as a running example in the next couple of slides I'm going to use uh this regular expression which essentially finds all of the URLs that correspond to uh Stanford and Berkeley domain so URLs corresponding to uh links from these particular domains in terms of interface suxing supports uh regular expression iies on both the flat unstructured interface where uh a regular expression search would yield all the matches in terms of the offsets within the input where these regular Expressions occur as well as the length of the matches and uh we also support the same operation for the key value interface where it would yield the set of keys that for which the values contain this particular regular expression so these are the two interfaces that we expose for regular expression searches and uh again in terms of evaluation if you were to look look at how exin fares against the same systems that I discussed before described before for the same uh setup and same data set and if you use the regular expression query that I mentioned before you can see that the latency on the y-axis is again in logarithmic scale and Spark uh both on disk and in memory achieves a very high latency and this is because in addition to having to scan the entire data set it also has to deal with the complexity of these regular expression queries which are far more complex in search queries elastic search on the other hand is able to achieve almost two orders of magnitude lower latency and this is because of uh the loose scen indexes that it uses as well as the libraries uh the regular expression libraries that have been highly optimized for these indexes now I would still point out that even though these uh uh regular expression libraries are very very optimized they still need to scan all of the index entries to materialize the results and in contrast snct actually executes these regular expression operators directly on compressed data without any data scan so you don't need to scan any index entries for this and therefore sixin is able to achieve an audit of magnitude uh performance gains even on top of elastic searge and this is uh for a scenario when the data already fits in memory for all the systems so these gains will only get much better for six synct if you were to P push much more data in memory so uh before concluding my discussion on succinct uh succinct spark package I'm going to briefly talk about I'm going to briefly mention that uh we've already been our techniques are already being used at lzv Labs they're using our spark package and I thought that the use case that they use it for might be relevant to the audience here today so I'm going to talk very briefly about it essentially their use case is to use it for annotation searches and what annotation searches boils down to is they have a huge collection of documents the folks at elier and these documents essentially journals Journal articles are uh paper Publications since they're essentially a publication company and in addition to these documents they also store annotations for each of these documents so given a particular document you annotate regions of text within the document with certain additional metadata a particular annotation entry might look something like the example shown over here you might have an annotation of typ type sentence or a noun phrase or a verb or a word and corresponding to The annotation you would store the region within the text where that the annotation corresponds to for instance in this example the first annotation is a sentence annotation that corresponds to uh character 0 through 15 in a particular document now uh annotation search actually involves complex analysis of both the set of documents as well as the set of annotations in tandem to Pro produce useful results uh and as an example this is actually an example that lzv Labs gave to us is uh we they wanted to find all all of the occurrences all of the sentences that talked about open problems in research in their papers now you can imagine why I'd be so I'd be so excited about this given that I'm a grad student I would love if someone could just tell me all the open research problems and I can pick uh whatever I want to work on but essentially this doing this is a very complex problem itself so the first step would essentially be to come up with a regular expression query that actually captures this expression of open research problems and this is uh their attempt of actually capturing this using a regular expression query which is essentially searching for uh you know patterns like remains unknown or is unclear or remains uncertain and the next step is essentially to find all of these Expressions within the set of documents now once you found these set of documents uh and the locations within these documents where these regular expression occurs you want to find all the annotations that correspond to sentences and contain these regular expression matches so you would actually have to check against all of the character ranges and see which ones actually contain uh direct expression uh entries so this essentially boils down to be a very very complex query and this is just one of the simplest queries that they have and what they essentially found by using six syn was uh they were able to reduce the storage footprint of these documents and annotations which can be huge to uh much much smaller storage footprint uh they had reported gains of almost 13x over the uncompressed representation and they can support all these qu iies at interactive latencies so most of these queries run at the order of seconds or even subsecs and uh essentially you could start using suin on spark today Yourself by just going to this URL and downloading the relevant packages and running it on spark if you have a spark setup already going and that's uh essentially all I'm going to talk about about suin on Apachi spark and at this point I'm going to switch gears a bit and talk about the second problem that I mentioned in the beginning of the talk don't worry I'm going to uh recapitulate the problem the problem that we were looking at was essentially uh you have distributor system that scales horizontally by partitioning the data across multiple servers and the load distribution across the different partitions in the system can be heavily skewed so you could have some partitions that have a lot of load and the others that are not so much and to put this in perspective of what I've talked about till now we've looked at succinct and how it allows us to execute a larger fraction of queries in main memory and now I'm talking about a problem where once the data is already in memory how do you cater to skewed query workloads and how do you even cater to situations where the uh skew across the different partitions may change over time so some partitions that are hot right now can become cold over time and vice versa as a concrete example you can consider the uh deployment at Facebook which uses a bunch of MySQL servers to store the C data and they store a bunch of MCD servers use a bunch of MCD servers that cash the hot data now the problem is that even in the M cachy servers that store the hot data in main memory there can be a huge amount of skew for instance you could have uh this arising due to you know really popular events like for instance Pokémon go being launched and everyone's playing Pokémon go and talking about it so you you have this becoming a very big phenomenon or for instance Donald Trump you know saying a fun statement and this becoming something very popular uh among the people at uh for for them to post about so this essentially leads to a huge skew in their M CD servers now since the data is already in main memory you can't use caching to improve your performance anymore so this is a real problem where a small set of servers uh see a huge amount of queries and are able to sustain that load on those small set of servers and a running example that I'm going to use for this particular examp uh for this particular problem is a system with 20 partitions uh distributed across 10 servers so you have two uh partitions per server for instance and we use a load that is zip fion distribution and uh the what we observe is essentially the load on the most loaded Shard is essentially 20 times higher than the least lower Shard and the traditional approach to cater to this problem is to use selective replication what you do in selective replication is essentially you selectively replicate certain uh partitions based on how much load that they have more concretely the number of replicas that you allocate for any particular partition is proportional to the load on that particular shot so if you have a partition that has much more load you'll allocate more replicas so that it can sustain the load now the main problem with this approach is that it's course grained and to see how consider partition the where the load on that particular partition is only slightly higher than what it can sustain so sometime something like 1.2 times higher than that the load that partition can sustain in selective replication you would allocate two partitions or two replicas for that partition and in a way your you're actually wasting 80% of the storage for the new Partition that you've allocated because that's not useful you only need to cater to 1.2x not 2x of the througho and this becomes a bigger problem when you have uh storage constrained uh situations or memory constrained situations where the partitions are already contending for the cash as much as possible so this leads to degradation of performance because you're wasting a space for some uh partitions that may not need it the solution that we come up with for catering to this problem is a system called Bluefish which Builds on top of sixing and essentially at a very high level if you have uh three partitions as shown in this particular example with loads uh shown as shown on the right so the purple Shard has the most amount of load while the blue Shard has the least amount of load what you essentially want to do is to transform these partitions change their storage Footprints in a way so that the load across these partitions become well balanced that's essentially the idea you want to pursue that would be the ideal case and what blowers strives to do is Achieve exactly this and to see how it actually achieves this what we do is we look at different schemes that exist today and we map their performance in terms of the throughput as a function of the storage uh that they require for instance if you recall uh the use of indexes uh require very very high storage but performes but provide high performance because they're able to speed up queries on the flip side you have scans which require storage but provide very high through uh but don't provide very high througho they have low througho and finally we talked about sucin which provide of which has an even lower storage requirement but uh achieves a throughput that's somewhere in between scans and indexes now what blofish essentially does is it builds on top of succinct and enables a smooth trade-off curve between storage and performance which means that you can uh increase or decrease your storage footprint based on what your uh performance needs are in fact what Blowfish does is it enables a dynamic navigation of the storage performance trade off curve which means that you can increase your storage footprint or decrease your storage footprint at very fine grand time scales based on what your performance needs would be as an example if you have a sudden spike in the load on a particular partition what you can essentially do is increase the storage footprint of that particular partition for a temporary duration so that it can cater to the increased uh amount of fluid at that partition and tying this in perspective of distributed system that scale horizontally by partitioning uh Blowfish enables the storage performance stateof curve for each partition in the system now in order to uh enable the storage performance tradeoff curve Blowfish introduces a new data structure called the layered sampled array now I'd like you to recall for a moment uh i' had mentioned uh several slides ago that s sync stores a sample suffix array where it doesn't store the entire suffix array but stores certain samples at regular location offsets within the array and in addition it stores a set of pointers that allow us to compute the unsampled values on the fly whenever required at slightly higher computational overhead now what Bluefish does is it takes the original sampled array in this particular example it has a sampling rate of two which means every second value is sampled and it decomposes into a hierarchy of layers and each of the layers have a different sampling rate for instance uh you could decompose this uh sampled array into uh three layers where the topmost layer has a sampling rate of eight where you store every eighth value instead of every second value the layer below it has a sampling rate of four where you store every fourth value except for values that have already been stored in layer eight above it and finally you have a layer with sampling rate two where you store every second value except for values that have been stored in the layers four and eight above it now if you stare at these two representations for a second you'll realize that they store exactly the same amount of data or store exactly the same data but two different representations what this new representation enables and uh that is the Crux behind Blowfish is that uh different combination of these layers leads to different points on the storage performance rate off curve that I had mentioned for instance if you keep only the topmost layer you have to store very few samples which means that you have a very low storage overhead but on the flip side you have to compute a lot more values on the Fly leading to higher computational overhead leading to low performance so in a way you can achieve low storage at the cost of low performance on the flip side if you stored all three layers you have to store a lot more samples but your performance is better because you have to uh compute much fewer values on the fly during query execution this means you can get much higher performance at the cost of higher storage footprint additionally what uh Blowfish also does is allows applications to dynamically increase or decrease the storage footprint by adding or deleting layers on the Fly using uh efficient addition and deletion algorith now having a dynamic storage performance trade-off curve introduces several problems of uh several uh new problems in a systems perspective for instance you could look at a single server in such a system where each of the partitions have different uh you know storage performance uh characteristics so how do you actually uh share the cach between different partitions on a local server now the solution that Blowfish uses is it maintains a request for each of the partitions on the server where each request queue simply stores the uh the collection of outstanding requests for that partition so if you have a lot of outstanding requests in a particular request queue this means that the load on that partition is very high and if you have very few uh requests outstanding in a particular request que this means your load is low and what Blowfish does is it maintains a low threshold so if a partition's q occupancy Falls below that low threshold it means that the load is insufficient and it deletes layers to free up some space on the flip side if you have U you know a particular uh partition where the u q occupancy grows beyond a certain high threshold you add more layers to cater to the high increased load but at the cost of higher storage overheads so essentially uh the initial goal we started out with where you want to be able to transform these shards or partitions in a way so that the their storage footprint is proportional to the amount of performance that they need is essentially what we achieve in Blowfish now there are a plethora of other concerns as well that we have to address for instance now that we can cater to this problem in a single machine scenario how do you share the global cache across multiple uh servers um across the different partitions that are there in the system moreover if you have a replicated system where each of the partitions themselves can have multiple replicas how do you share the cash across the different uh part replicas and how do you schedule requests across these different replicas that may have different performance characteristics now without going into much details again uh Blowfish uses a unified solution to cater to both of these challenges and the solution is to use back pressure style scheduling what back pressure style scheduling enables essentially is for us to continue sharing the cache across the different partitions proportional to the load on these partitions but uh we do so in a distributed setting without requiring any explicit coordination so we are saving a lot more on the coordination part of this particular scheme now I I know I've skipped a lot of details here as well and I'm happy to talk about these offline if you're interested but I'm going to take a step back and talk about the performance so as compared to the state-of-the-art that you have in in terms of catering to skew that is selective replication Blowfish achieves 1. five times higher throughput for the uh running example scenario that I had mentioned a couple of slides ago that is you have if you have 10 servers with 20 uh shards we are able to achieve 1.5 Times Higher through put than selected replication and what's even more interesting is that our performance for this particular scenario is within 11% of the optimal throughput that you could achieve for uh this particular setting uh now I'm going to conclude my talk by just giving uh you a brief Uh current status for all to the two of these projects we have a stand loone system implementation that uh contains all of the techniques that I described for six synct as well as Blowfish which we have prototyped as well as tested and it's constantly evolving based on new feature requests as well as improving the performance we also have a spark package for which contains all of the suin techniques uh and allows analytics on compressed rdds and uh finally we've also uh open sourced all of the different succinct and Blowfish techniques in three different languages in C++ Java and Scala and we've introduced these in three different languages because of ease of integration so you might have applications that might might use uh these libraries and whatever language that they're actually implemented in so that's essentially all I have for you today and I'd be very happy to take any feedback and any questions that you might have about s sync and Blowfish and thank you so [Applause] much yeah short question so isct something that's build from to of spark or is it just presented it as some combination with right so essentially uh succinct question yeah so the the question is whether succinct is uh part of spark or did I just present it in a way that it seemed like it was a part of spark so uh in truth synct is a standalone uh technique in itself collection of technique in itself which essentially enables queries to be executed on compressed data and we have a standard loone system that is exists out outside of spark that performs all of these complex queries on a system that supports high throughput and low latency however the spark package that we have focuses only on the analytics aspect of it so you could perform something like text analytics on spark and that's why I focused on uh the spark pack today as a yeah probably very simple question but suppose you had like stock market data coming in and you wanted to focus on a range of time like December 5th December 5th and then look at say count all the simple symol say for C certain stock can you first execute a query that will get you the offsets into it and then have a secondary query that will only get the count with within in that range definitely so the question is uh given that you have a stream of data coming in such as in stock markets which have associated with them some sort of uh time identifier and you want to perform a range-based operation where you want to look at only a certain window of time for those uh you know for the data within and you want to perform queries specifically for that time window can you actually combine a range based query together with uh a more complex analysis such as say search on that time window and the answer is definitely yes so we do support range queries and uh search queries individually and all of these queries can actually be composed into perform more complex operations and uh the use case that I mentioned earlier that is The annotation search actually just combines one of these examples is one of the examples that we can uh you know support and definitely the stock Market example would work with the succinct use case as well thank you yeah okay and uh if you have no other questions that's all I have for you today thank you oh yes com so that's a good question um there are several components to that question given uh you know that you're just asking about the Wikipedia data set we essentially partition the data and uh I think we had about 40 partitions for it and each of the partitions was uh roughly about a gigabyte and our uh compression scheme currently supports compression at uh I think 16 somewhere between 16 to 32 GB per hour per core but it is highly paralyzable so you can essentially use multiple cores to speed up the compression techniques and uh uh so essentially if you wanted you know exact numbers for how much time for it took for us to compress that particular uh data set uh for 1 GB of data I think it took about 11/16th of an hour to compress it yeah and that is only using one core you could speed it up using multiple cores okay that's essentially all I have for you today then thank you so [Applause] [Music] much