sfspark.org: Alexander Ulanov, A Scalable Implementation of Deep Learning on Spark

sfspark.org: Alexander Ulanov, A Scalable Implementation of Deep Learning on Spark

Recording: sfspark.org: Alexander Ulanov, A Scalable Implementation of Deep Learning on Spark

hi everyone my name is Alexander thank you for coming I'm going to talk about a scalable implementation of deep learning on spark it is a joint work with Hungary mank from data bricks very driven Bush from hallway quoc anh Li from spark community and there is minovsky from hewlett-packard clips four of us didn't know each other until we met on the spark developers mailing list and spark actually was a platform which enabled us to collaborate and to do some good development and research on a common platform that actually beneficial for everyone the outline of my talk is as follows I'm going to talk about the basics of artificial neural networks then I'm going to focus on the implementation of multi-layer perceptron in spark and how we did optimization and fertilization I will conclude with experiments and future work and since this talk is based on a talk on spark summit on the recent sport summit I also list here what is new comparing to my previous talk so first of all I added some slides about their presentation heuristic that I'm going to talk about and also i did some new experiment with larger cluster and also there is a new slide design actually now it's huge pocket enterprise probably you know that they should be split into two companies so now I'm in the green one so artificial neural networks it is a statistical model that approximates a function of multiple inputs it consists of interconnected neurons and the exchange messages niran produces an output by applying a transformation function on its inputs and urine Cyril groups grouped into layers and network with more than three layers of neurons can be called deep can be an instance of deep learning and on the right hand side you can see a small example of neural network which has two inputs three neurons in the hidden layer and one output there are different types of layers and layers is defined by a transformation function the most common is a thin transformation when you multiply the input by some weights and then add bias and simo it which transforms the input into the interval 0 1 which might be very useful for classification purposes you see the shape of the function on the right hand side also there are such layers as convolutional soft mugs etc multi-layer perceptron is a network with several layers of a fin and sigmoid layers and model parameters are the weights that neurons use for transformations and those parameters are iteratively estimated with the back propagation algorithm and multi-layer perceptron currently is widely used for speech recognition in particular for phonemic classification and in computer vision a set a player of the network and multi-layer perceptron was released with spark 11.5 so it is in the main distribution and you can use it already so first of all let me show you an simple example how to use it in spark so assume that you want to build a very simple classification classificatory suffice the input images for that classifies the handwritten images into digits 0-9 so it's actually a very famous data set golden East it consists of the grayscale images of 29 28 x 28 pixels and they represent digits so then we need to build a multi-layer perceptron with 784 inputs is the number of pixels and 10 outputs is the number of different digits that we want to crucify and also we can add two hidden layers that would provide better accuracy as suggested in this paper and the network would look like on the picture so it has seven hundred eighty eighty four inputs and two hidden layers and then ten output to do this in spark you need to do the following so first line it loads the data set second line creates a classifier and sets the layers and the block size I'm going to talk about it later and then the last line it trains the model and after that you can use this model for prediction and the similar thing you can do in Python so basically the same three lines of code so it's very simple you can also use multi-layer perceptron in spark pipelines like for example on this screen we also load the data set on in the first line then we create a pc transfer mater that extracts 20 features from the data set and then we use those 20 features for classification so it is the third line after that we create a pipeline of pca and multi-layer perceptron and train the model and the same can be done in fightin so basically it is to show that multi-layer perceptron supports sparked by planes which are quite useful so now let me talk about the implementation itself and the problems that we faced so we addressed three main requirements in our implementation first of all we we conform to spark spark api's then we wanted to provide extended exchangeable interface so that other people can use it and also the perceptron should be efficient and scalable just it should be it should be fast so why conform to spark API because spark can call any Java Python a scholar library not necessarily designed for spark actually you can use any Java or Python library that has multi-layer perceptron and call it from spark but this approach has some problems so first of all its results would expensive data movement from spark to the library and second it prohibits from using this library in spark machine learning pipelines so we had we addressed this requirement second it should have extensible interface so since park is open source product then probably other people would like to contribute and would like to write new layers or new functions and features so we had to write extendable interface and easy-to-use interface for developers and our implementation processes each layer as a black box in back propagation in general form so that it is easy to implement new layers and features so we basically need to implement rather simple interface and currently there are several features and the development in particular convolutional neural networks as text out encoder a restricted Boltzmann machines and others the last requirements requirement was about efficiency and scalability the efficiency was addressed by batch processing as i mentioned multi-layer perceptron consists of several layers several pairs of Athene and sigmoid transformations and a thin transformation can be represented in vector form as follows so it's basically a multiplication of it's a multiplication of ways by the input and addition of some beers and it can be represented in this way like on a picture but vector matrix multiplications are not as efficient as matters matters and then we can do batching so we can stack as input vectors into batch and perform mattresses multiplication and that means that the variables and the question they just become matrices with corresponding dimensions and it can be represented on this picture so it's basically instead of multiplying matrix by vector u x mattress and in this way you can gain performance and it's very simple so we implemented batch processing in matrix form and that enabled the use of optimized native blast libraries I will talk about them on the next slide and we also did some optimization so that we limit the GC overhead so we try to reuse the memory all the time so what is bloss blass is basic linear algebra subprograms it is a hardware optimized natives in c in fortran there are several famous implementations like Intel MKL open blast which is usually comes with the Linux distributions also there is an interface for GPU called envy Blass and it's different from Buddha bless because kudu Blass is not a standard Blass implementation so it's a little bit tricky and in spark we can use blast through the net lib java library which is bundled with spark and to use the binaries we need to compile spark with specific flag but is also very simple on the on the screen i'm showing the performance of the matrices multiplication with the use of different blast libraries so on the y-axis there are seconds and it's a terrific and on the x-axis are the matrices sizes and with different colors i show the different libraries and it's obvious that there is a huge benefit from native glass comparing pure java f to j plus so if you want to use some linear algebra you probably should compile spark with specific flag and use native blast so you can see that the dark gray line is kind of very slow but it's multiplies measures is in a very long time and surprisingly GPU is faster only for large matrices it is the light green line and it is due to the fact that when we use any blood which is an interface for GPU blood we do a copy from the drm to the memory of the GPU and when the computation then the actual multiplication is takes a long time then this transfer from memory to memory it seems to be slow it seems to be little comparing to the computation and if the computation is very simple then you just spend a lot of time on transferring and that's why for and that's why GPU is them only good here for large matrices but still you can gain some performance also there are more details on this site and in our recent paper so it has more experiments and also on the right hand side you can see the hardware configuration that I have used and it's basically not the freshest hardware that you can find it's not the most expensive one ah ok so with this one ok ok ok ok so oh yeah now I hear it yeah actually I thought that this microphone works on cool so I hope that you did not keep the previous flights also the fact is that indeed you can take advantage of native glass and spark and you should do it if you do some linear algebra and in particular matrix multiplication the last part of our requirements is scalability so as I mentioned we addressed efficiency by using the budget purchasing and using native glass and scalability is more tricky thing as I mentioned before the parameters of the model are estimated with the back propagation algorithm and this algorithm is iterative so we have to do plenty of iterations and on each iteration you update your model and I'm going to briefly explain this process so on each iteration each node gets parameters from master so in a single note case it would be just one executor that gets perimeter and that's represented on the left-hand side so those parameters are the weights of the model then on the next step each executor computes gradient given the data that it has locally and that might take some time after that it sends the each each executor sends the gradient to master and at the end Master computes new version of parameters based on the gradients and it give it it can be done in different ways and there are different types of gradients actually like for example batch gradient when you process all data on each iteration stochastic when you process random point and mini batch when you purchase random bunch of data and I'm going to talk about the batch gradient one of the reasons is that stochastic is very hard to provide it's basically a sequential algorithm but we are also looking into it so the next question is how many workers to use so it seems that if I have less workers then we can we can do less compute obviously but if you have more workers then we have more communication because we send the parameters on each iteration and we can have millions of iterations and we can just defeat the purpose of more workers by spending a lot of time and communication so there is a communication computation trades of here obviously and let's let me talk about it in details so assume that we have a data set with g data points f features and K classes and we want to train a logistic regression for simplicity it has FK parameters which is features x classes and we can write the formula for communication and for computation in this case so poor communication we have n workers and they get and receive FK 64-bit parameters so it's double precision we can have FK 32d parameters if you have single precision so they they send those parameters through the network with a particular bandwidth B which is usually one gigabit or 10 gigabit and some software overhead see which which is just software overhead that is connected to your software that you use and in case if you use all reduce pattern for communication you will end up with the following formula for communication for so in this formula two stands for the fact that you need to get and then two and then to send the parameters so initiative on each duration you do to communications then 64 FK / b is just the is just the number of seconds that you need to transfer one set of parameters and sees our overhead logarithm stands for the OL reduce communication pattern so if you have used simple communication pattern when you send it when you send parameters at one time as i have shown on the previous slide then you would have n instead of logarithm which is worse then for computation part each worker has p flops so it has a CPU and the CPU has developed and the worker should protest d divided by n bit of data so i assume that the data is equally distributed over all workers we have an workers ng date so each one has d divided by n data and it needs to do to FK operations and it's basically a multiplication of matrices and vector measures of parameters and vector of the input and based on these estimations we can write a formula for the computation so it's d divided by n you paralyzed by data on your workers and two if k divided by p is just a number of seconds that you need to spend on computation so it's rather simple formals and then we can try to estimate what is the optimal number of workers for a particular data and for a particular model so we can just minimize the sum of those two times and if you minimize it you will come up with such formula and we can even generalize it for any model so this formula is for logistic regression and the last formula is for any model so i just replaced to FK which is the number of iterations i replaced it with L so L is number of floating-point operations for the particular model and then based on this formula you can estimate the optimal number of workers by putting the actual numbers of your data set and the number of ways in your model and the number of computations you need to do and also the network speed and the number of flops so let's take a closer look on this formula and see it if it makes sense so more flops means lower degree of parallelism it is simple because it's in denominator so the more computations you can do then the more obvious is the communication trade-off like you compute it very fast and then you need to spend a lot of time on the network so that makes sense then if you have many iterations so the model is very complicated like deep learning you have many features and a lot of classes and a lot of parameters that means higher degree of parallelism so you spend a lot of time on compute and then network time is not the big comparing to the compute time and if you have a small overhead for sending or getting a message it also see means higher degree of realism just because you save time so it seems that it make make sense and I can give an example for the particular data set and a particular cluster so there is a data set called missed eight million it's also hundreds in digital cognition data set it has what 8.1 million of documents also 784 features 10 classes and we want to train a logistic regression I have a very slow cluster with 32 gigaflops double precision CPUs and soul network with one gigabit and overhead of 0.1 second which is just an estimation of the overhead and spark and if you put the numbers in the formula you will get that the optimal size of the cluster for training this model is 12 which also seems reasonable for artificial neural network we can also estimate the number of operations needed for each iteration and this slide just shows how you can estimate it like how to estimate the L for your model so for the multi-layer perceptron for forward pass you will need to W computations for bed / Quebec propagation to WF and for gradient also to W and totally it will be 6 w and then you can put a 6 w instead of foul so I'm not going to go into the detail but it's the show that you can estimate the number of operations for your model and then use this number in the formula to get the optimal number of cluster optimal size of the cluster in some cases you may not have this optimal number of workers like you may not have like 12 nodes in the cluster 100 nodes whatever it will show but it seems that it is okay because if you plot the graph of this formula then you can see that it's after a quick drop the time decreases slowly so on the y-axis is the time I need it for one iteration and on the x-axis it is the number of nodes in the cluster so when you have only few notes adding one more node actually increases decreases the time a lot because the graph drops but then it starts decreasing slowly actually and adding more nodes does not actually provide you a very big speed up and at some point even it starts to increase if you look at on the right hand side of this graph it is because the trade-off of computation and communication you will have so many notes that they will congest the network and you'll spend a lot of time on communication so it seems that we can use less notes than the optimal number and still get good good performance still get good speed up and we can also estimate what is there how how slower is my cluster comparing to the optimal one or if I want to be only ten times only ten percent slower than the optimal case how many nodes should I have in the cluster and it's also easy to estimate using this formula so on the left-hand side it is the time of your cluster on the right hand side it is the time of the optimal cluster and after some computations and a little bit tricky approximation we will get the formula for how much is our configuration slower than the optimal which is K here on the slide and also how many nodes should I heaven in my cluster to be K times slower than the optimal and a cure is an example so we computed that the optimal number for logistic regression is 12 and how many nodes can I have to be only take ten percent slower it is nine and i think it's it's nice result because you can reduce your cluster by three nodes but the performance will be slower only ten percent and it's it may it might be a good trade-off in terms of money and speed so after all these elaborations on the scalability heuristic we did the actual scalability testing of the multi-layer perceptron that we implemented so we used again this minis data set with 60k samples and we used very deep network so actually not very deep but deep network with six layers it contains 12 million of parameters and it has also taken from one of the papers on this about this data set we use rather good cpu with the 100 gigaflops double precision so it is a cluster of of the nodes with such CPUs and we compared our implementation with cafe which is de planning to from Berkeley and it is a single note implementation it can use GPU and it's it's extremely fast and for GPU we use Tesla also not not a very new GPU but still what what would head for spark so for cafe we could use only one node and for spark we could use many nodes and actually on the slide we used 13 notes so it's not the case net five workers that is written here so actually we use more so on the right hands yeah so so this is the network with six layers and each layer here has this number of neurons but each neuron has more parameters because it has a lot of inputs also each yes yes yes yes so basically to compute the number of parameters you need to multiply 784 by 200 2500 then plus two thousand five hundred times 2000s yeah so that's how much later perception works as I supposed to convolutional net works for example okay so on the on the right hand side there is a graph that compares the speed of cafe and our implementation for this particular network and on this particular hardware and different colors they represent different different things actually so there the green color represents the total time for one iteration of the multi-layer perceptron and it depends on the number of nodes that you use and you can see that the shape of the graph is similar to the theoretical graph that we had one on one of the previous slides so on in some point in time you just stuck with the a lot of communication and the time doesn't decrease anymore with the number of nodes and it seems that there is a optimal number of nodes here so like around five or six and if you add more nodes you would not gain a lot of speed the orange line stands for cafe so which is it is a simple note implementation and the orange line is CPU implementation and you can see that for one node we our implementation is slower than cafe but I think it's a kid because our implementation is in Scala and there are also some spark of our heads so we are not dead worse than cafe on single note so only 1.7 times slower but if if we add more nodes we can scale preaching pretty pretty good like for example 5 nodes give us a 4.7 XP dub and it actually beats cafe on a single node obviously it also comes close to GPU but only in the matter of computations because you can paralyze computations but then you will have a very big network overhead and like on the right-hand side of the picture and it will just defeat the purpose so that's kind of is consistent with our theoretical formulas that there is a trade-off between computation communication and probably there is a optional number of node and you can just use this number without spending much effort on a big cluster and actually I can I can also put those numbers of our experiment into this formula and if i do it i will get 15 so formula tells me that 15 number of nodes is optimal for this particular experiment yes it's not the case probably here and this is probably connect connected to the fact that spark doesn't use all the reduced communication pattern it has a different one which is not logarithmic so probably were in between of linear and with a rhythmic and if we use the second formula which is needed to estimate how much slower I you then the optimal case it gives us 1.5 so it shows that with five nodes we are only y 1.5 times slower then in indicates with 15 nodes so we have a cluster that is three times less than the optimal and the performance is just one point one point five times over which is also I think a useful result that shows that you can see we use reasonable size of cluster and you don't have to use it a very big one and you can try to estimate what is the optimal number okay so let me make some conclusions conclusions and talk about the future work so we have implemented this multi-layer perceptron and it's available in this park since 1.5 so it's not the package is included in to spark code and if you use spark and then you can use multi-layer perceptron it has API in Scala in Python and it supports pipelines so it is easy to use we also provide extendable in terminal API and further contributions are very welcome so it would be good if people will write new layers and somehow advance the state of the art one of our conclusions is that native glass and also GPU can speed up spark and if you use some linear algebra computations then probably you should use native blood within spark and we provide some useful heuristics for fertilization of buzz gradient which allows to estimate which should be the number of nodes new cluster for for the fastest computation currently there are several new features in progress in particular staked out in further actually it's finished it's not yet emerged into the master branch restricted Boltzmann machines it's also done but it's not merged dropout it's in progress and coloration on neural networks is it's done but actually currently they are optimizing performance and for the future work we are look we look into several different directions so first of all we try to come up with some new heuristics how to paralyze the batch gradient and we look into adaptive lb of GS and it's kind of interesting interesting thing because I supposed to a stochastic gradient it has better conversions it's usually converges faster but the convergence is proportional to the number of documents in the training set and if you have very large training set then it takes a long time even that convergence is fast so we look into ways how to have a different convergence rate ratio and also look into the stochastic gradient and parameters server so it is another approach how to provide the gradient descent and the synchronous version of parameter server is not right now is not feasible in spark so it should be a separate component and it's kind of another line of research and also we continue work on you and there are many people working on them and I invite you if you're interested also to contribute and it's very interesting I think so thank you very much for listening and we have time for questions