Scale By The Bay 2020: Rohit Jain & Bin Fan, Optimizing Latency-Sensitive Queries for Presto...
[Music] we'll be talking about how we solved latency sensitive queries at facebook with our collaboration with alexia so let me start by talking about presto at facebook scale first presto is a query processing engine uh it's uh it's takes equally into input and does etl and analytics related jobs for the users presta is pretty big in facebook it has roughly around 40 000 servers uh pretty much every day in facebook it scans roughly around one extra bytes of data most of these jobs are etl in nature um data scientists are writing their pipelines scanning data using pasta or writing using faster and there is a lot of ad hoc analysis done which is done as well using presto and this ad hoc analysis and dashboards are very latency sensitive queries and those were the challenges which we were looking to solve with alexis so this is a very brief architecture diagram of cluster by no means it is complete it does highlights the important parts which we want to focus on for this particular task so let's look into this so we get the sequel and we have a component called planner and optimizer which takes the sequel passes it tries to plan create different plans for it and optimizes it so the planner and optimizer in order to able to its job it needs various partitions and files information so that it can create distributed plan and give it to multiple workers festo roughly has two main components one is called coordinator another is called multiple workers so the current in current architecture of cluster is that it has like one coordinator doing well doing the planning related things and the workers are responsible for actually doing the real work so the planner and optimizer uh gets the table and partition related information from hive metastar and for those partitions it accesses the hdfs which is which is something equal and we use in facebook and gets like the file block information from uh hdfs for those partitions and then the scheduler basically after looking into these files it distribute this work for the logical plan it needs to execute to these workers the workers interact with this remote data source which is hdfs in this particular case to open the files and read blocks from this workers in the presto are compute only nodes they do not really store any data with them so which means like uh the only truth of truth the only data source which is in the hdfs which is in the remote location while this design allows us to scale because we can add as many workers as we want in the cluster at any point uh it becomes a problem as well because every time relying on a remote source can be a problem from the latency perspective and which is something we saw while like this particular architecture was allowing us to grow uh as fast as we wanted if we saw like oh we need like more load to handle for particular queries or at the peak time we can add more workers into it and similarly the other way around but as i mentioned like this particular state always relying on the remote data can uh can be a big impact can can have a big impact in the query latencies okay so let's look into like how we solve this problem so one of the common optimization technique uh when handling with remote data sources is to cache the working data set cache the working data set closer to the compute node and the idea is fairly simple uh basically like you are making lesser trips to the remote storage for the data can help out not only with io but latencies as well okay so this is a little more complicated diagram uh with our rear architecture again it doesn't really tells the whole picture but it all highlights the components uh which are important for this particular topic um so if you notice like the worker communication with hdfs we have a local ssd in between this local ssd is available with every worker node and it is storing some part of the data along with it so that is the idea like if you are accessing uh any data from the htfs if the data is not available in that cache which is this local ssds uh we will cache it so the first preference would be going to the cache um so this is a read-only cache so once so the data request so this is the flow looks like so the data request comes for the worker worker looks into the ssd if data is available it reads it back if the data is not available into the ssd like the cache is supposed to read the data from the sdfs and return it back there are a few more concepts uh like soft affinity is defined in and the idea is behind the soft affinities basically like uh let's assume like a coordinator has around 600 worker notes which it is working with uh these different workers might have different data set cached into it so we wanted to create some kind of affinity to make sure uh we leveraged the same worker again and again for the same data set for which we are executing so how does this affinity gets created so we talked about uh before distributing the plant to the workers we get the file information from the hdfs so based on these files or multiple splits gets created a file may be a pretty big file the file gets broken down into different splits and for every split this plan along with the logical plan gets assigned to the worker so these jobs uh for the for the spirit we try to make sure like uh we have some affinity uh so these splits have some kind of affinity with the workers and this is what we call affinity work so before this new architecture we used to have a random node scheduler random is probably not the right word but the idea here what it is trying to convey is the coordinator it basically distribute the no distribute the workload randomly amongst worker obviously the workload is taken into the consideration to make sure like no worker is overused and all the workers gets used for the workload but with and but this kind of scheduling would not have been helpful if we wanted to build some kind of a data caching because if a same split for two different jobs guess gets assigned to two different workers then the same data might get cashed into two workers and that would not be the best use of the space available because this space is not unlimited it's limited uh disk space available in the ssds we want to leverage the same worker again and again for the same split so we implemented to start with something very simple what it was like we just take the split and whatever the path uh in the htfs is available for that particular split we just like uh take them we just divide it by like the number of nodes take the reminder of it and assign that particular uh node so as you can see like the algorithm in this particular case what we started with is fairly simple like the split the reminder choose the node give it to the same thing we obviously take like the workload of a node uh into the consideration if the worker has got too many splits already processing we choose the different uh the secondary second so the first one we call like the preferred worker the second one we call it the secondary secondary preferred worker but even if the secondary is also fairly busy processing under the node we choose like the random node or the least busy node which is available and in that particular case we do not uh leverage the data caching this algorithm is fairly nice and we can see like this can be easily broken with the addition of a node or a node being lost so we have already implemented the consistent function to make sure the node lost do not affect the data caching hit rates okay so when we started working on this problem uh we came up with a bunch of options uh facebook internally has lots of great caching libraries which we could have leveraged we had few open source solutions and we started thinking about like should we build our own um so the facebook internal libraries are great they have lots of great functionalities but some of them did not work for us mainly because they and we didn't really want the caching as a service we wanted the caching to be invited to make sure like it doesn't really get it with latest extra latencies and the another important point about this facebook internal caching libraries was like they were not really an open source okay and best of being like the open source we had great affinity with the building solutions which are available to the open source world anyways so we started with uh building something of our own which was a pretty nice solution we wanted to get started quickly and roll quickly so we built something pretty name what it was doing like it was just copying uh the warm storage files to the local storage as as soon as it was copying it was also merging these uh files to make sure like the file count remains uh limited into the local disk um we tried out our use cases with uh these things and as you can see like our cash flow trade was all over the place it was like up and down all over the place what we realized with this is uh that like because we were merging these local files into a bigger chunk every time like the eviction event happened uh we realized like we were evicting a large chunk of data and we were not really finding reusable bits and pieces from these big files so this is the point we understood more about our use cases and we learned about alexio and we started collaborating with them so what we realized like uh we wanted like a java based open source library uh we wanted segment based caching what the segment based caching what i mean by this we wanted data to be read right and debate in the smaller portion or smaller blocks not in terms of the bigger file units because as we saw earlier the eviction can evict a lot more data than it should uh with the bigger file evictions uh we wanted like pluggable eviction policies we wanted like different configurations uh we wanted various stats for the cache usage so that we can understand uh how is the cache performing what kind of issues we are dealing with uh with various failures the software and the hardwares we wanted to make sure like the caching doesn't become a single point of failure which means like even if there are some issues going on with that caching the operations must continue by ignoring cache and reading from the remote source we wanted most of these operations to be asynchronous so that do not become a problem from the resource perspective uh we wanted like better file management we also wanted like the file throughput limiter to make sure we avoid the flashing difference issues so one of the thing which i wanted to mention like so when we started working with alexa alexa was available as a service which was not our preference as we wanted to work with an embedded library so the cloud data is the lecture we mentioned uh what we are looking for let's say folks what you open for it and we work together to carve out a library which can be labeled in the facebook world so performance well so we tried out like one of our use cases uh and we tried out like two days worth of full queries from the production cluster uh and we shared it to the shadow cluster we roughly tried out around 17 000 queries we tried it out on the 600 node list each node was configured with 460 gb of cache we started leveraging lru eviction policies and we did with 1mb block size which means we were always reading writing and evicting in one mb size so these were the r performance results as we can see like we saw remarkable improvement all across the numbers our p95 improved by close to 50 percent similarly with p75 these are the latencies times in terms of millisecond um so some of our queries which was taking like 30 seconds they started taking 14 seconds and similarly io savings this was like a huge win from the i o savings perspective which means uh for the data size where we were reading 582 terabytes of data we now were reading only 250 terabytes of data so we roughly had around 57 percent scan savings for this particular use case the cache trade which is uh fairly reflective of the latencies improvement which we saw it was fairly consistent and as you can see like it remained fairly uh almost picture perfect uh there were some tips uh which reflects uh of the queries uh which basically like uh are rare uh which doesn't really happen regularly and they scan the data which was not really cached so which means like there is some query which happens very frequently very not very frequently but it disrupts the whole cache so this also tells about like some of the further improvements we need to do is like build some kind of histogram and where we understand what kind of passion what kind of queries might uh be the disturbing disruptive from the cache perspective uh to further talk on the lecture deep dive i will hand over this to bin uh to talk about yeah thanks for hit uh i will talk about some part of the luxu focusing on the elixir part for this collaboration as you may or may not know a luxia is open source data oxidation system which is a new layer sitting between the compute and the persistent storage in the data stack alexa is commonly used to speed up the olap queries for sql engines like presto hive spark as well as like ai or machine learning jobs for engines like a tensorflow or pi torch next slide please yeah a little bit more about alexa history it's open source data oxford system and it's originally uh originated from uc berkeley and lab as the research project named the tai cam was open sourced in 2014 the top contributors created this company and to continue the support and grow and commercialize this open source technology backed by the top investors like anderson harrods today the open source project has more than 1 000 contributors on github all over the world next slide alexa has been widely adopted by hundreds of different companies across different sectors all over the world in production including financial services to retail or technology to internet content so you can see a lot of big names big logos on this site as shown in this figure yeah next slide please so the way people use a luxury often is to deploy it co-located with or close to the compute processes but with a different storage either on premise or in a public cloud mounted into the luxury namespace in this way a user can basically treat aloxi as a data abstraction layer you think you are dealing with a logical file system and applications are only accessed this logical the path logical path in this logical file system without being aware of the real backhand physical storage like you don't need to worry about which client to use which storage uh parameters to use which credential all this can be handled by the admin setup together like so the developers and admins can just use this freely without really understanding the details this can greatly help simplify the development and decouple the evolution on both application side and storage worded data topology so it helps also ease the maintenance of data platform in addition to the performance next slide so for the collaboration with facebook we also originally as mentioned is a full feature distributed system but for this collaboration i will focus on the development with facebook to integrate presta with a local library version of microsoft embedded into private workers so this luxury local cache provides a hd hdfs compatible interface to press the workers and this hdfs interface is backed by either both the external storage and the local directory to store the cache data on the cache hit it will just fetch data from local cache storage or on the cache miss this caching file system will just go through this uh go to the external file system to fetch data and put this data back into the local cache storage next slide so here are some technical details when implementing this uh local version embedded version of luxury cache particularly informed by presto workloads running at facebook for example first in order to work with the non-sequential read pattern which is commonly seen reportedly commonly seen for queries in facebook we choose to use the fixed size segments as we call pages to uh data in the auxiliary space and we also store them off heap rather than in jvm on heap storage to reduce java g gc pressure secondly because presto server is highly concurrent by design we have to implement a highly concurrent and thread safe locking scheme to prevent uh to provide price to workloads high performance but also stress safety next slide in addition uh next slide please yeah in addition we also introduce a plugable cache policies including including lru as a default and error view as optional this cache library supports different storage engines include a store simply using a local file system as the storage engine but also a more advanced storage engine based on rocksdb we also provide a async cache right feature to handle to help press to handle birthday cash inserts so this can support async rights isn't a cash rights which cue the cash inserts in the background and run this in the background also because we're running on facebook scale the disks are we found it's in almost deterministically to run into certain problems either hardware defects or the file system issues it's basically deterministic you can not assume the file system call to the disk will succeed or return you a failure signature uh like a in the bounded time no there's no such guarantee it's just because simply because the large scale were running so we have to deal with this carefully to provide the failure recovery scheme either during this cash transaction or during like a startup time next pipe so there is a lot of ongoing work so we have already deployed alexia with presto at the large scale and it's running pretty well in production for quite a few months right now in facebook people are very happy about results but there's a lot of ongoing developments for example to support uh the different level quota cash quota across the uh in presto so with the cash we can just tell the cache oh spend no more than one terabytes for this particular table or for this particular scheme schema we're also as mentioned by rawhid we see the dip in performance for certain cases and we we found a lot often times it's due to uh it's caching a lot of small files and we're doing a lot of performance optimization for the small files and for the future work we're also discussing the semantics that we're caching for different files can we do something differently next slide yeah to use a luxury local cache with presto it's pretty easy you can see these three lines of change in the configuration press the site and also we have a documentation we have a blog documenting this for all the technical details you can also go to this link to read our blog next slide this is my last slide if you are looking for so the embedded version of luxury cache is is basically providing you the basic functionality but there is a lot if you are looking for a full feature caching control distributed caching system and this is something called a luxury file system you can look for including fine grain control on the working side you can free and they can ping data in the cache or you can metadata caching and syncing automatic automate this process if there's any change on your backhand so you can just invalidate the cache in alexa space and you can trigger the data transformation service so you can convert certain formats into some other formats more friendly to sql engines like presto or you can trigger some migration service to move data from one storage to another while providing no difference to the applications from the application they are still only accessing this logical files logical directories and we also provide a lot of centralized file system command lines so we can easily deal with the data in the cache for more details you can read this documentation i listed below okay so that's the end of my talk i'm happy to take questions yes hey hey javin uh so we have a couple of questions uh um uh here on the audience um how do we manage memory sharing between dram and persistent memory asks uh i guess this is a question for me on the more on the luxia side uh so for aluxio it's uh it's interesting that we have we have the we have the concept like we're running we're storing data on the of chip storage but we actually we actually we actually put the like running unluxia in like using jvm memory right that's the that's the dram but you can use alux to manage off-heap storage which can be either dram or persistent memory or a combination of dram and persistent memory or some like ssds for rocket you can you can check me but like i know facebook they're using large volume ssds to uh for alexa to manage as a caching space so that's basically separation like the code runs on jvm uh using dram but you can for the odd data it's off heap it's not in the jvm but you can you can combine directories from different system memory and dram or the combination it's a it's basically up to your choice yeah so uh yeah that was the i guess the next question was like do you deal with uh zone storage uh um i think as as long as so there there is uh two answers to this so first um i guess like we need to understand better like what's the difference in the zone storage uh this audience is particularly interested in but as long as you can integrate with unlocks or enter storage we have a persistent android storage interface as long as you can dealing with integrated or compatible with that then you can just write a plugin or use existing plugin to talk to our storage uh that's that's no problem and and if you're talking about multiple different zooms and as long as they are not migrating data to each other and you can just treat them as a different storage and mount them into a luxury space uh as just back in different storage directories logs to logical storage directories or if you want to have enable more advanced logic across like a different zone storage that might be harder but like we can i can i'm happy to take this offline to understand the request better you