scale.bythebay.io: Feynman Liang, Deep distributed decision trees on Apache Spark
Recording: scale.bythebay.io: Feynman Liang, Deep distributed decision trees on Apache Spark
so today I'll be talking about deep distributed decision trees on Apache spark specifically I'll be talking about a package that we released called Yggdrasil this work was part of an academic research project it spans numerous universities here all my collaborators we presented this work last nips sniff is actually coming up not like three or four weeks in SoCal but we presented this and Barcelona 2016 so who's all familiar with decision trees let me get a show of hands sweet awesome I didn't include any backgrounds and it looks like I don't need to so we all know that deep decisions or decision trees themselves are really good for a lot of different kinds of machine learning problems they're highly interpretive or easy to train easy to tune and you can just plop one out show it to a business user and ask does this make sense however scaling them is a whole different story as the data set size which I'm going to denote end throughout this talk grows or is the dimensionality P grows we see a lot of different scaling issues such as the data not fitting on a single machine or as the tradition tree depth D gets really large we see an exponential number of leaves so the question of that we're going to answer during this talk is how do we train deep decision trees on a distributed computing environment most of the most of the most common approach is known as planet it's what's currently implemented in SPARC so if you call the random forest a classification random forest regression this is the algorithm that will be run the data isn't fit on a single machine so we take every individual instance here denoted by a row and we partition them across our cluster and then on each worker they have a subset of data on which they try out all the different possible candidates splits a candidate split is good if it results in more purity so all the like +1 labels going to one side minus one label going to the other side and then all at once this compete computation is done on all the subsets its aggregated back onto the master so the master can figure out how well did this candidate split do on my entire data set there's some problems with this approach however namely if I want to compute all possible candidates splits for all possible candidate features then that requires n times P where n is the number of instant sorry P is the number of features and times the number of partitions number of round trips and furthermore if your features are continuous so they're not categorical then there are like l minus 1 so in between every two data points you could split and so that results in a lot of communication problems and this is actually what the authors of planet's cited in their paper it's a little bit hard to read but it says planet makes a trade-off between finding the perfect splits for an ordered attribute and simple data partitioning and in particular prior to tree reduction we wrote a MapReduce and compute approximate eques depth histograms for every order attribute why do they do that well it's the problem that we are talking about with a continuous feature if you have n different data points and it's continuous there's n minus 1 possible spots you could split on and so you're basically shoveling the data set back and forth between your master and your worker on every possible candidate split and so the heuristic proposed by planet is let's compute histograms and instead of trying to compute every possible candidate split let's just consider B of them where B is much much less than n this saves us on communication but the trade-off is if B isn't one of the best splits we're never going to find the best split because we never consider it there's some other problems with planet however with this introduction of the B parameter you have yet another knob you can fiddle with on your machine learning model you have a trade-off between finding the optimal split as well as the communication cost and the runtime efficiency of the algorithm and finally we'll see in the asymptotic so that is as the dimensionality P or as the data set D gets large or the depth of the tree D gets large we're gonna see really poor run time even when the optimal peak be split number of Bin's is chosen here's our proposed approach and it's actually not too hard the only difference well there's a couple differences but the key distinction to note is rather than partitioning by row we're partitioning by column so we're not partitioning individual instances we're partitioning individual features this is reminiscent of columnar databases and it's kind of what we stole the idea from rather than having each worker compute the sufficient statistics how well did this split do on their subset of data they sense a split only involves a single feature they compute all the possible splits for a feature and they figure out the single best candidate split for that feature aggregating back to the master what are the best splits for this feature subset that each workers responsible on notice here there's no more approximate binning we don't have this parameter B because it doesn't matter that we consider all possible candidate splits it's also equivalent to a serial algorithm on a single machine we don't have to implement any form of locking to ensure that the entire data set has been processed and we can now start actually aggregating the summary statistics you can take this and write a serialize linear execution for every possible parallel execution of this algorithm the way that we do the column partitioning is extremely simple in our spark package there's a method called wrote a column store dense and an equivalent one for sparse data sets called wrote a column store sparse all it does is your takes your RDD of a vector so your wrote your rows and it transposes them note however this is actually a very expensive computation it requires reshuffling of all of your data set and so if you were to factor this into the entire algorithms run time it wouldn't be too efficient but you'll see that many data sets are oftentimes just stored in a column or formats for easy aggregation easy analytics use cases so here's an animation of how your soul looks we start by taking all of our features suppose we have year built square footage price per square foot and then we partition them across who this is kind of messed up okay let's unfortunate what you would have saw was you would have seen these features partitions so half of them would go to worker one half of them would go to work or two and then the workers sort each of the parts each of the features by their value so now all the features we've shown here are in sorted value and we've colored their labels with green and blue and now we compute the best split for each feature this is super duper easy all you do is you start on the left hand side you slide that threshold to the right updating some summary statistics each time and then at the end you go over all the plot things you considered and you pick the best one you pick the best feature and the best feature split you send that back to the master so then then the master selects from amongst all of the responses returned by the worker the single best feature to split on for this iteration of the algorithm so let's look at these two things side by side on the left here you got planet you got partitioning by data instance and on the right here you got your saw you have partitioning by feature the algorithms are very very similar the only distinction is instead of sending back summary statistics here you're sending back the single best split that I already computed on this worker and instead of computing the split here that's that gives you the highest purity or information gain over here you're just picking the best split because all you're sending back is the best split there's less hyper parameters you no longer need a proximate binning to deal with communication the communication cost is morale is very similar you're missing this constant right here but this two to the D is inevitable because that's the size of the tree you're shoving back and forth across the network and you're picking up this interesting term right here which may be alarming at first notice that and all of a sudden showed up in our communication a cost are we shove it are we now communicating the entire data set that's not very scalable well answer that in the later slide and finally rather than considering only be possible splits we're considering every possible split so we're guaranteed to find the optimal split for each feature and now if you look at certain and now if you take these asymptotics and you start showing regimes on which which dominates the other you'll see that for high dimensional data as well as for like very very deep trees Yggdrasil has better asymptotic runtimes than the planet approach but this isn't news like I said we still call them partitioning from the data from the databases literature and column column based partitioning has been done many times in the past how did we get this published in nips this is like this this is an this was another paper published last year in 2016 and they say that this process of partitioning by column is very communicational e efficient however the repartition of the data will induce a very high communication cost and the reason why that's the case is because after you've done a split the nodes have either gone left or right and so now in the sub net now in those two leaf nodes if you wanted to do another split you should only consider the subset of the of the data that went into that leaf node so not only do you need to compute you do need to communicate the path that each net that each data instance took down your tree and that's that's the origin of the N term in our communication cost the way that we mitigate this problem is by being very efficient in our communication there so I've highlighted in green the key distinction that'll of that that we use to to mitigate this cost and keep it minimal specifically rather than rather than compute whether than sending around each nodes entire split path we send around a bit vector from each of the workers back to the master the bit vector is a 1 if the node went to the right on that split and a 0 if the node went to the if that instance went to the left on that split the animation isn't there there's the there's the bit vector and then afterwards the master will select the single best split out of all of them compute this bit vector for the entire data set and then send it back to the workers so now that each worker will know ok the data that I have on all the data that I have it's split like this now after this next iteration of the algorithm and then this font this finally this last part workers sort each feature by bit vector then by value so when you're in the next inter and you only want to consider splitting on the nodes that went left by sorting on the bit vector notice that now we have two ascending sequences rather than one I think this says price but you see there's an ascending sequence here corresponding to the sorted left subtree and then an ascending sequence here corresponding it to the sorted right subtree when you're choosing the candidates split you should consider the information gained on this side and this side independently because those nodes won't fall into the same branch of the tree when your algorithm is run here's the code that is the entirety of what I just described you have a transpose transposition phase or you don't need to do that if you use a column or data store for each iteration you compute the best splits and gains this is done this is done in a parallel manner on each of the worker instances these best splits and their information gains are then aggregated back to the master along with the splitting information so the aggregate bit vector the bit vector of the zeros and ones which direction each node went once this is a grenaded we then update the partitions with the best split that we found so that the next iteration of the algorithm can continue forwards and we do this until a fixed number of a depth D or after that you say the trees too deep you don't want to train anymore why do we do this sorting well because because of the splitting we have to keep track of the the splitting trajectory of each of the training instances if we were to instead just copy then we'd have a copy into a new column after every split rather than doing the sorting on our existing data set we'd have an exponential increase in the memory use utilization it'd be it would be multi exponential to the number to the depth or the number of training iterations instead when you sort you're reusing your existing memory and so the footprint of your memory is constant furthermore computing the best split only requires a single scan and most importantly sorting based on the bit vector is Big O of n rather than Big O of n log in as you might expect this is a little bit surprising and to see why this is the case it's important to remember in the previous iteration of the algorithm we have already either ordered by the feature value or sorted first by the feature value are sorted first by that bit vector and then by the feature value so we can actually leverage the ordering that we got from the prior is from the prior iteration of the algorithm to sort the next iteration of the algorithm in linear time the way this works is for every single one of the nodes in the tree what I'm gonna do is I'm gonna get the range that that node of that tree is responsible for and then I'm gonna iterate over it and I'm gonna grab the bit so whether it's split left or right from the aggregate bit vector and if it's a left I copy it into a temporary array for the left if the right I copy it into a temporary on the right and at the very end I just merge them together it's kind of like merged it's it's almost analogous to merge sort but going the other way some other additional optimizations we do is cross is sparse bit vectors because the bit vectors are zeros and ones and if you're datasets balanced you would expect that to be about roughly 1/2 1/2 you have 50% sparsity and so compress run length encoding can help you pretty far we can also encode the labels and finally we think and we can compress the feature columns so we can apply schemes such as Delta encoding or run length encoding on the features themselves this is another idea we stole from databases the ability to Train to do analytics on compressed data specifically in the pre-processing of our algorithm we modify it now to sort on the call sort on the values in the column as before and then to compress it and then never decompress it but rather to continue operating on this compressed column representation why can we do that well recall we already need to sort the features so after sorting the features you start getting these runs if you have repeated values of any individual feature and then these feature values are visited in sequential order when determining the best split remember when I said we start on the very left for the threshold and then we just kind of marched that forward to the right computing the information gains for each split this kind of sequential traversal pattern is precisely the reason why we can stay using run length encoding of the future columns the downside however is after run length encoding you can't just shuffle up the intermediate values the way that we had our little linear time sorting trick code and so you'll have to maintain an additional space to keep track of the splitting trajectories of each of the data instances but the upside is immense cache locality we'll see later that feature compression results in much smaller data sets that will fit on memory no diet no DRAM access and that you'll see an overall improved improvement in run time on a variety of data sets so let's get into the results how does this actually perform in the real world so we benchmarked on the very popular m n-- estate millionth and compared to ml lib when you're training a trie depth of 18 layers Yggdrasil performs six times better than ml of will citrate XG boost but it was timing out on our spark cluster after around 1313 layers deep we don't think XG boost was for deep trees actually so that's reasonable on a dataset that Yahoo provide provided which consisted of two million instances on a regression task we see similar scaling as predicted by theory when the tree depth gets large ml libs training time grows exponentially whereas we see linear scaling with Yggdrasil x' training time there's a 24x increase when you're training trees of depth 18 and finally we measured the amount of but we measure the communication cost so the number of bytes sent across the network when running these two algorithms and we see that in almost all training regimes well the scaling is are similar in almost all training regimes uracil is your just shell is transferring less data as measured in bytes across the network we also see that your gazelle is more scalable when you have higher and higher dimensional data sets and you bump the tree depth from 13 to 15 you'll notice that ml live goes from this solid this solid red line down here all the way up to that line up there whereas for you Giselle it goes from this solid dotted purple line just up to that dotted red purple line you go she'll is both more scalable is more scalable in terms of both the number of features as well as in the depth of the tree and finally those little nitty-gritty optimizations that made us different then column our data stores it actually does matter you can see that we've dropped for the performance for in training time by over 40% just by applying these optimizations and that each one adds on top of the previous one it's available as a spark package it's already compatible with spark 1.6 and if your codes are ready in spark ml lib all you got to do is change decision tree classifier to your sole classifier and remove this max Bin's parameter because you don't need B we're gonna try all the splits there's future there's upcoming work to merge this into spark 2.1 we're planning on adding features to let to let the algorithm choose automatically whether you want to run your girsl or whether you want to run planet we're also looking for to add approximate binning to further cut down our computation costs and we're also adding support for when a single feature does not fit on a node that's all I got for you thank you so much for your attention do you have any questions [Applause] you didn't about the quality of the models this Queen precision accuracy yeah that's correct we didn't report any empirical results on the quality and the reason being is if you have a layer 10 a 10 layered decision tree trained using the same rule you would expect more or less the same tree modulo the binning and so in our nips paper we reported the results on quality and we saw that the quality was more or less the same however for certain data sets depth actually didn't matter on the quality and so if you can so a tree of depth 5 like trained a lot worse than a tree of say depth 12 [Music] [Applause]