Scale By The Bay 2018: Alex Sergeev, Distributed Deep Learning with Horovod
Recording: Scale By The Bay 2018: Alex Sergeev, Distributed Deep Learning with Horovod
you Alex remover from machine learning platform our job is to make everybody successful deep learning and be able to train models in hours instead of months so why people talk about deep learning these days this is famous slide from Andrew ng and basically what it shows is as you continue to grow size of data that the organization have traditional machine learning algorithms start to lose your performance start to kind of become constant and so as you naturally grow more data you don't get further benefits and this is where deep learning is helpful in a sense that it can continue growing your accuracy or metrics that you care about which can turn into business value for you there is few right now deep learning is quite it's a state of the art for many debates it's state of the art for vision for machine translation for sound processing and so it's very natural for us as a company to be able to leverage some of the new research that happens in the community and that's why deep learning is great and we use it for a lot of things so we use it for self-driving cars obviously because there's a lot of image processing video processing light or processing control systems we use it for trip forecasting where we want to know okay it's getting me next new year how many people will try to drive uber or write uber what's going to be distribution across the city how we can predict that fraud detection we want to make sure that everybody is doing the IR honest on the platform and not trying to cheat and so that way we can maximize earnings for drivers and actually minimize price for riders and there is many more use cases that I cannot cover here so how does it work it's pretty simple the reason the flurry is called deep learning is because when you do the your data prediction you essentially pass your data through a series of consecutive transformations and where transformations you do the deeper is initially what was called artificial neural networks before would typically use one or two of those transformations and now it became deep learning because now you do hundreds thousands of those transformations and so when you do training you typically what you do is you pass if you do supervised training you'll pass data and label to your model so label is your expected output for a given data and data goes to the whole model you get prediction which could be right or could be wrong in case if it's wrong you see that you made a mistake and you want to correct your weights in a way that your model will become more close to model that will correct predict this correct label for this data in future and so what you do is you compute these gradients going backwards in the model so you basically first do this forward pass/fail model then you do backward pass and afterward you did the whole both of those then you can start next iteration with your training so as I mentioned you get most benefits from deep learning when you have more data and so when you have a lot of data that makes things slow because your machines can only ingest so much data the second GPUs can only train so many images per second and we need to have some solution to still be able to train these big models one a lot of data in reasonable time so engineers can have mentality of quick iteration because for engineers it's pretty easy to get no state that okay I'm going to kick off this training it's going to take a week so I need to do something else maybe that with twiddle my thumbs or think of some theoretical things or I'm gonna spin up 10,000 hyper parameter searches trying to search this space and see what's gonna happen frequently what happens is they did spin up 10,000 hyper parameter searches but then they all of them converge to nothing because there was some fundamental issue that you they found after that week and so being able to train models quicker maybe using same amount of resources but quicker every time allows you to get to the right model much with way less iterations and so if you think about these big models in Big Data there you could think of either distributing most machines were building models so big that it doesn't fit in a single machine what we see right now is that modern hardware has good capacity in terms of memory if you build a double years you have 120 gig of GPU memory same and GCP if you buy fancy DJ x2 machine you get 500 gig of GPU memory which fits almost all the models that you can dream of unless your Google and so because of that biggest recipes to basically use data parallel training and so the way their power training works is you have these copies of the model that you pass the data through one is the same data is different you computed gradients in the penalty then you are edge gradients and then you do next situation and so there are a lot of ways how to do this data parallel training all the frameworks have slightly different flavor it UX is good in some frameworks it's worse than other frameworks and we see that a lot of researchers actually flip between frameworks quite a bit because let's say Google or Facebook or other company publish new paper and you want to try this paper this approach on your data so what you do is you take their code from github you try to apply to their idea that unit is big so you need to do this review training you need to figure out ok how do i distribute this framework how do this with that framework on my system and so with our platform team we had kind of two goals how we can enable people to be successful so on people one goal is for infrastructure people like me we need to choose servers we need to choose network hardware we need to choose container environment we need to choose default containers for people to use if they don't want to build their own wanna tune distribute training performance of the whole system on like system level and I want Emily years to not worry about any of that technology years don't need to be systems people they don't know how the studio training works they just need to build great models using frameworks that they love which may change on databases and so that's why I beat corrupt so horah what is a library for distributing deploying frameworks it works with data flow TAF Kara's by torij and we have all going contribution for other frameworks as well and it's really easy to add another framework if you have something that you really love and it's not here it installs on top of frameworks by simple people install so you install latest version of the floor or whatever version you like whatever version of biters and then you install 10 qira'at you don't need to have special version or anything special with the framework it uses advanced algorithms that I'm going to cover a little bit later in the presentation and can leverage wide range of networking technologies that may make it perform way better than it would otherwise and it's separate concerns between different actors so infrastructure people can set up environment needed for it to work great and engineer and machine learning engine years can just train the models make super simple notification and go from training a single GPU to train your thousand GPUs so how does it work primary driving factor in hora what is algorithm called ring or reduce that I'm gonna cover pretty quickly basically the idea here is this example we have three workers each of them have six numbers and the goal is an end to have average or in this example some in every position across all three workers so what ring will reduce does is it partitions those buffers on each worker into number of segments that is equal to number of workers and there's two phases of algorithm in the first phase you send data from where every worker and receive data from one worker and also add numbers and in that of this first phase in one of the positions in the buffer you will have the right answer for that column and the second second portion of the algorithm does distribute that value so what happens is it follows the amount of steps that you do in this algorithm is proportional to number of workers but amount of data that you sent is not proportional to the number of workers almost constant on average it's closer to the total number of data you have per worker the reason is we actually send very small offers and so if your initial buffer is very large you can see that the never iteration will actually sound pretty small amount of data and total summation of data will be constant it will be just function of the size rather than million of workers that you may have this holds if your buffer is reasonably big because if your buffer is too small and you have million workers it may be hard to divide let's say 1 megabyte into million pieces and send one byte is not going to be more effective than send one kilobyte because it's still the same packet but for large bars like gigabytes buffers it holds pretty well there is paper by these fine people who promoted this algorithm and they claim that it's bad with optimal for this reason so Horowitz tag as I mentioned it's installs on top of tensorflow it uses MPI for worker coordination the reason we did worker coordination is because tons of flow does perform operation in a non-deterministic way so when I mentioned here we do this layer computation when we compute these gradients here it's actually one worker could compute this gradient first and this gradient the other worker could do it other way so we use coordination to make sure that we resolve any conflicts between these workers and we use nickel which is Nvidia optimized or reduce and other collective communication library for actual very fast or reduce so how do we use it it's completely quite simple so first to initialize the library you just import it and do HP dated you pin GPU to be used so tensorflow if you just run it by default likes to use all the GPUs can find in the system we typically run horowitz in a way that you run single GPU per process and so because of that we need to specify which GPU you should use for every process so in this example what we see is that when we do MPI run on these two hosts each host has to reveals it shows kind of two concepts that we have in Apia one is called Frank another is local rank so rank is universal idea of the of your process in the whole world of this one whereas local rank is rank within a server and so it's very natural use GPU that is identified by local rank and we provide primitive to get that local rank and plummet to telephone another thing that you need to do is you need to use hormone distribute optimizer on top of your optimizer that you like to use it can be anything it can be momentum as 2d Adam pretty much anything and you need to adjust learning rate so adjusting learning rate is one of the interesting aspects and it's open research space there is quick and dirty rules like Facebook popularized a linear scaling rule so basically what you do is you multiply your single GPU learning rate by the number of GPUs that you have if you have a lot of GPUs let's say in their paper they used 256 DP use you also need to do a little bit of warm-up so in first few iterations you don't start with a large learning rate you start with small and then over first couple of a box you'll scale that learning rate in ers we provide callback for you to just include in your code that will automatically do this warm-up but it's not hard to do in pretty much any framework and in our examples you can find that callback and adapted to your needs an important part is synchronized initial State so when you launch your training let's say you didn't start from checkpoint so starting from scratch you were workers we will start with some random set of weights and so if you average gradients across those weights they intestine this example may point not where you want them to point so they all point to the center individually but not collectively and so to avoid that what we do is we pick one of the workers as Golden State and we replicate that state to everyone using broadcast typically we broadcast from a first worker and so there is two examples how to do it in tensor flow there is a lot of api's that you can use our goal is to work with pretty much an API that you like to use and so here we have two examples one using winter training session another using just normal tea obsession how do you do this broadcast another piece that is important it's simple but sometimes overlooked is how do you do checkpointing so because all your processes run exactly the same code they all by default route right checkpoint to the disk sometimes you run multiple you have multiple GPUs on your system then you would write multiple checkpoints typically in the same file that will not end well and so recommendation is to just write checkpoint on the first rank because all of them will have exactly the same or no state so it doesn't make sense to write everywhere you don't really need to solve this problem you just save it in one on one rank and you're good and so if we take all this together this is full example how you would do your training of your model and the model is hidden in these three dots so we're not talking about how to define models we're just talking how to scale them which is pretty concise and so we talked about tensorflow but there is more so this is how you do tensorflow but we have quieres on fighters and one of our goals as I said before was to make it really easy for users to switch frameworks and so this is here is full example how you do the same stuff and carrots and as you can see it's very much the same you still pin GPU you still change your learning rate you still write that you miser and you still do the callback to do the broadcasting so names are little bit different but concepts are all exactly the same same forests America estimated guy is one of the new areas the tons flow release also so they use this estimator you still do pretty much the same set of steps and the API is look pretty much exactly the same and as one of the more recent additions won't set at bihter support because a lot of our data scientist like to switch back and forth and even in Python we managed to preserve pretty much the same API you still specify GPU you specify optimizer wrapping used to broadcast parameters you the parameters here are also a little bit different but you know the steps that you need to do and it's just a matter of kind of looking them up in a syntax how to do it so how do you do how do you run horror typically use MPI and so if you just use open API one of the latest open you need to do let's say this execute this command which what it will do is we'll run four processes on current machine so let's say you have four GPUs then it will run for processes if you have four machines with four GPUs you can run this command so there is some brittle code that you always need to put and protip is for your environment you just create this boilerplate wrapper wrapper script that will hide all this from user so user will not run actually this it will he will run MPI run dot SH and just his command and what it does you basically you as a infrastructure engineer I can define what kind of network interfaces you use in your environment what kind of other flags you may need to set to get best performance but you were machine lyric engineers don't have to deal with any of that and so one of the recent additions that is currently in pull request but you are very welcome to try it out today is hormonal spark so on spark what you can do is you don't need to do in Karen instead you can define your training here as a function and I completely avoided pretty much any code in the actual model but imagine you have put the examples that I've shown before in this function and so then what you can do is you can start your training just using this horrible spark run and what it will do it will start your training on your spark cluster you may do it on CPUs if your model is small enough or you may set up your spark cluster on GPU machines and then just do a you spark for both data processing and data training and how would you do training loops so one of the ideas here is that you can actually use spark for validation native spark rdd's to execute validation and you can use this horrible training for actually running training runs so this way you can hide complexity of launching impaired run it allows user to leverage existing spark infrastructure that many people have in their company you can do data preparation and terrain in the same environment so oftentimes you do a lot of data extraction data massage before you actually do training and in these environments this helps quite a bit and so one of the Pro tips is for reading data inside this training function you can actually use pittosporum and so what you use is you save your RDD into parky format and then use Pettus term which is different library that we recently open source that allows you to do randomized reading from barca files so typically doing randomized reading from something in SPARC is quite a bit hard and pedis term is optimized exactly for this scenario and integrates quite well so hopefully this all this talk before convinced you that it's reasonably easy to use and so now let's talk about performance what we observe in our environment is that we get really good scaling with Horwood so we scale to 512 GPUs and we got around 90% skill efficiency for common models like Inception and rest net we did get around I would say C 70% with vdg and VT is really hard to scale the reason is VG has a lot of weights and it's very short it's very easy to compute back and forth but amount of data that you need to sell this quite a bit big it's actually I think bigger than rested and so because of that many frameworks have way worse Callum efficiency but we still see that some of the models they still use fully connected weights which are root cause of these big layers so it works pretty well on this virus as well so how do you get most of the hormones so let's say you start running you see some performance your EQ is getting improve this performance so there is some tuning knobs that you can try to adjust and see if it helps you one of the recent additions that we did with help of nvidia and amazon is hierarchical reduce this works on really large environments but also works in some of the smaller environments under certain situations so what it does instead of doing one single ring will reduce it does hierarchical reduce that I will describe a little more detail in just a second one other thing that we have is tensor fusion tell of you was there right from the beginning and what it does is it actually batches small or reduce operations so if you think about rest net let's say progressed at 155 it has 155 layers and it has around 500 tensors a lot of those tensors are really super small and so what you want is you don't want to run all reduce and all the coordination for every little tensor so what we do is we batch those smaller reducers until we get let's say 65 64 megabytes which is this default fusion threshold that we see and we do this analysis every 5 milliseconds which is the cycle time so you can actually adjust both of those parameters and see what works better for your model on your environment and finally we have added fp6 you know reduce so what it does is instead of doing all reduce using full precision weights like a P there are two weights what it does is it you can still train your model in the parity to but before doing or reduce it will cast your weights down to F P 16 dual reduce and then cast it back so what we found is two things first when you do a P 16 training it's actually quite challenging because some of the layers they tend to have very different weight distribution and actuation distribution and there is techniques that you need to apply to actually make it work there is something called mixed precision training that if you're interested in Vidya has written up quite a bit of information about what we found about all reduced though is that it actually is way less sensitive to this reduced precision and so when we did this very naive experiments without any rescale in just trying ResNet training we see that we have no changes in accuracy when we do this physics in training but what we saw if you we run it in less than ideal environment we see up to 15 percent or 70 percent speed-up so it's something to try it's not gay and like these previous things it's not guaranteed that it will not change your model outcome so it's something that you need to opt in but it can give you better performance and so argue or reduce is quite interesting so I wanted to spend a little bit time okay so basically it's similar situation to before now we have four processes and two machines and each of them has four numbers so what we do is it's three stage algorithm in the first stage we do so code reduce scatter when we use it using nickel so for a reduce scatter on the first two digits we will reduce on the first half of the of the tensor and it ends up on the first try first rank then we will reduce second half it ends up on the second machine same for the other server so after that we get ground truth for with a host or but it's spread around then what we do is we do parallel or reduce for rings that are formed around local cid so for all the first GPUs we'll have one ring against each other well the second abuse will have a ring for each other and so on so forth that way you do local all gather in that so this algorithm allowed us to actually scale quite well all the large systems we do have users from national lab laboratories like Oak Ridge where they have 27,000 reviews and algorithms like that help them scale way better than just single wrangle reduce so I also want to talk a little bit about horrible future and how we're looking about making it here to you so this chart shows you change how changes in these parameters that I talked told you affect outcomes so a blue line here is nickel and the green light line here is hierarchical reduce and so you see that typically they're they behave differently in some cases one wins and other cases other wins and it also depends on the cycle time it depends on the fusion buffer which is on top so it's kind of hard and time-consuming for you to find this best value and so what we're doing right now is we're in patient optimization water tuning which will try various combinations during first few dozens of iterations and after that you will not have to worry about finding right the nation were evaluating whether this can find this right combination for you and so far we see that in most instances it does and other improvements is further gradient compression so we see in FB 16 we're looking at using for a bit we're looking into using gradient compression with stochastic gradient compression which is allows you to save up to ten thousand times or 1000 times some bandwidth so for example sometimes you'll come and ask okay I have this cluster of GPUs I have one gigabit network between them right now we say sorry you cannot do anything with this but with things like gradient compression we could and it's right now it's active area of research so my last slide is what did we get with it and how do we use it so in uber hora what is the only way to do distribute deploring we train convolutional net works less TMS and other models we train them and hours instead of weeks that it used to take before hora what is used by other companies it's used by Amazon and VJ Alibaba and various research institutions quite a bit it's included in various different distributions so all the cloud providers AWS TCP after they have horrible in their default VMs a video review cloud IBM has it in their fabric for deep learning which is another open source project that they released they have it in their Watson Watson studio and in greater breakfront so it's quite a bit used we'll be really happy if you guys find it useful for your cases please let us know please let us know if it works for you or if it doesn't work for you and if it doesn't how we can make it better and then I'll say thank you [Applause] thanks very much chance so I'm sure there are some questions so when you have like several models of different users so you update the gradients you average them within BG but how do you update the model do you like average all the gradients afterwards and then update the model or you do like set of micro mini batches so right so the question was we have all these models and how do we update gradients against them so what we do is we here we see three blue boxes in reality is let's say you have thousand GPUs there will be thousand boxes and whenever you do training whenever you compute gradient we start all reduced so like averaging of this gradient across all thousand models so the string is thousands appearing now a thousand it may already starting to be more reasonable to use this hierarchical algorithm but this one operation of averaging covers the whole universe sure everything crashes mpi and nickel are not fault-tolerant they just crash so what typically you do is you actually write checkpoints quite frequently and when you crash you restart from not checkpoint typically when you think about these environments you actually think more of HPC environment where you have this like very expensive servers if you look at list price of DD x1 or d j-- x2 they're quite expensive so you have this expensive servers which are not supposed to fail very frequently and so and you'll also use way less of them than if you do it let's say spark MapReduce job so it we don't see that much issues with failures in a single job there happen sometimes but they happen rare enough for you to be able to just restart from a last check point and not spend too much money or time you