Distributed Graph Processing with Scala and Akka

funconf 2013, Adelbert Chang: Distributed Graph Processing with Scala and Akka

Recording: funconf 2013, Adelbert Chang: Distributed Graph Processing with Scala and Akka

uh so today I'm going to be talking about uh distributed graph processing with skull and AKO uh please let me know if you can't hear me or anything uh I'm running on zero hours of sleep I hadn't anticipated there being a hackathon at my company last night so I'm running on four or five cups of coffee yeah uh so let's get started a little bit about me uh I'm a fourth year student at UC Santa Barbara where I study in the bsms joint program in computer science I also I'm also a research assistant on campus where I study uh large scale graph Mining and modeling and uh cluster Computing which is how what I'm going to talk to you guys about today sort of came about uh this summer I'm an engineering analytics intern at box where uh I am using Scooby I'm doing some stuff uh some M produce jobs to do some machine learning and and my abstraction of choice is Scooby I play with sculling a bit ended up going with Scooby mostly because I like the abstraction better uh I've been using Scola since January 2012 uh I got into it purely because we needed a distributed graph processing system and at the time I knew Python and C++ python was too slow for us and I would rather not write a distributed system in C++ so I picked up scol and so that's what we're going to talk about today uh an outline I'm going to talk about the motivation for why I built it a little bit about the context and assumptions the system was designed specifically for my lab uh our lab has a very specific set of requirements and I wanted to build a system that made our job as easy as possible I'm going to talk about the user and system requirements I had for a system uh the solution uh show you a lot of code and then hopefully assuming nothing goes wrong I'll show a little live demo of of the system running on a lab cluster so the motivation let me pull up these notes higher okay so the motivation uh a lot of our algorithms were or are embarrassingly parallel and so Google does have this preal model for graph computation but it has this uh different view on things where it says you think like a vertex and you think at each superstep I'm going to receive messages from other nodes and I have to think about what messages I'm going to send to other nodes and how all of that adds up and while that would probably work for us uh we wanted something simpler uh we wanted to just be able to write the algorithm tell it what work uh what sort of input the algorithm took and then have magic happen and have it distributed so for example uh in shortest paths uh for an unweighted graph what we would do is just give each worker one note at a time and then we would just run a simple breath for search and then return a mapping from uh a destination node and the shortest path so we did a want to go through the trouble of thinking okay at this step I'm going to send this to to the other nodes I just wanted to write a BFS and then uh distribute the work based on uh at at a node level granularity uh some context and exceptions uh specific to our lab we are studying large scale static graphs and these graphs are usually from online social networks so you can think of each person being a node in the graph and then a friendship or a follow uh being a link between two people we have a cluster of around 30 machines the cluster shares a file system it's I believe it is Network file system uh graphs are very large uh they span they go anywhere between a few thousand nodes to a couple million that's probably tiny compared to what Facebook deals with but it's still it's not a trivial amount of nodes but our machines have enough memory to fit the entire graph into memory so we didn't want to keep doing some fancy uh loading and dumping to this we just wanted to keep it in memory and compute on that uh and we also wanted Raw results dump to disc so for for example uh if we wanted to find a degree of all nodes for uh or something we wouldn't reduce it down to like the average degree we would literally just dump two disk a file mapping a node and a sree and we would just have one massive file and then after that we would take it into something like Python and then uh do whatever analysis we wanted on that raw data some user requirements I had for a system uh the users should not have to deal with AA uh I was building this system with some PhD uh for some PhD students to use as well and they didn't have much experience with schola and I didn't want them having to worry about oh I have to first spawn this master actor and then worry about I to spawn these worker actors anything like that it should have been really easy for them to get a job running they should only have to define the algorithm input so like I said earlier uh just Divine a BFS uh and Define that I want the work distributed on a perod granularity they should be able to put an upper bound on the number of threads per machine uh we don't really have a schedule in a cluster as of now the way we sort of work is uh maybe a P student a student right now is doing some data analysis on this machine and say okay I'm going to only deploy maybe uh two worker threads on that machine and that frees up some threads uh for for the other students to do analysis on so the system requirements we had uh we should be able to deploy without any previous cluster setup we have NFS we didn't want to have to install hdfs or dup or Hive or anything like that any uh complex configuration we needed to be fall tolerant of course uh a worker died the whole machine the whole system shouldn't go down we want to be elastic meaning uh if at some point in the middle of the computation some machine freed up or we added a new machine to our cluster or whatever I should be able to just uh spawn a worker on that machine point it at the master and have uh have work distributed to that new worker without any having to shut down the the the master or anything graph should be loaded locally uh what this meant and this was probably obvious but the very first iteration of the system I built when I had about 2 month months of scol in me was that I had made the uh messages sent to the remote actors uh such that the graph was packaged with each node and that was obviously a complete waste of network resources at this point because all the all the machines are are sitting on NFS together they should just be able to load their own copy of the graph and keep it in memory and then the only thing that would be distributed around is the work so like a per node or a pair of nodes or whatever and the whole system should clean up after itself uh there shouldn't be any just no like Hadoop job track or anything uh we are launching this on a per job basis if you want to do average degree then we spawn one single instance if we want to do uh pag rank we spawn up with separate instance no uh Central job coordinator uh the inspiration for this uh there was really good timing we I was designing the second iteration of the system around uh summer of last year I believe and that was when AA the AA team was holding its summer of blog competition where people were supposed to submit blogs uh in regards to AA doing something cool with AA submit a Blog if they liked it they published it and this guy called Dereck Wyatt had published a post called balancing workloads across nodes with AA 2 and this was precisely what I wanted to do with my system I wanted to have a Central master balance workloads across several workers and I had it's very simple to balance because the work was all the same I was each worker would get a note finish get another note so on and so I wrote a follow-up post to that called Distributing and memory graph processing with AA which basically took his system which was more of a general description of a generic Master worker system and I added some few things and changed a few things to make it specific to uh my use case which is in the context of graphs uh so basically his plug post and what we wanted was uh he described the idea of a balancing dispatcher right so AKA comes with this idea of a balancing dispatcher for local machines where you have like the central mailbox backed by several actors and then when you send a message to a mailbox that mailbox that proxy mailbox would handle sending that message to an idle worker so that's basically what we wanted as messages came in it would Target only the idle workers as the workers finished new messages would be sent to them if there were still messages to be sent uh yeah talked about this so yeah this is basically how it works right we have a balancing dispatcher uh we have a mailbox as workers as workers are idle messages get sent to them once they're working uh if new messages come in it will either sit in the mailbox or wait until a different some other actor becomes idle once the actor becomes idle they get the new message so on and so forth uh we wanted this at the remote level so across several machines in our cluster and so with with the actor model this is really simple right uh basically we have a single Master which represents a dispatcher that stores all the messages inside uh some sort of like a queue or or a list or something each remote actor or the worker has its own mailbox and then it will once it's idle it will send a message to the master telling it that I need more work the master will give it work it will begin Computing and once it's done it will it will know when it's when it itself is done and they will send a master saying okay I'm done Computing give me more work and so this becomes really this becomes a very easy implementation in terms of uh the actor model uh some design decision that I had to make was so AKA is capable of doing remote deployment and remote lookup uh the benefit of remote deployment is that uh if the master deploys the worker the master by definition now knows that the worker exists uh remote lookup the benefit of that is that the downside is the master doesn't know the worker exists until the worker registers itself with the master by sending in a message but the positive side to this is that uh like I discussed earlier when we had added a new machine to a cluster or or machine freed up I can just spawn a new worker and point at the master and so if we did deployment once the master had spawn the worker actors there was no way to there's no easy way to get it to spawn more actors on other machines and so I decided to go with the remote lookup method and that actually uh worked out really good so basically just how it works we have the master the workers will register with the will send a message to your master at that point the master has the actor reference to that remote actor it will save it inside some sort of queue in uh in its internal State and then from there on it can do the whole balancing dispatcher thing uh the overall design of the system is pretty simple we have the outside world sending in some sort of work to the master the master will queue up the work uh as workers spawn it will tell the master it's idle the master will send the worker some work uh the worker will go into some sort of busy State once it's done it will uh trigger some sort of event to make it send another message back to the master Master will notice it's Idol if there's more work it'll send it back to a worker so on and so forth and so now I'm going to start showing some code uh basically I'm going to show a majority of the code it takes to implement the master and the worker and then I'll show a full application using uh what this what I call this uh the name of the system I call it saber I'll show a full application of saber and then hopefully I can show a live demo and uh and then we can finish off so these are the messages that the master receives right so from the outside world uh it will send a message to distribute work containing the work that it wants to distribute so maybe like a single node or a pair of nodes uh there's another message that the outside world consent to it called all work sent that basically tells the master uh I'm done with the work and that's how the master knows that once all the workers are idle and all the work is sent and our worker queue is empty then we can begin to clean everything up uh a worker can also send messages to the master of course uh you get the worker created message that's how the that's how the master that's how the workers tell the master of his existence and that's when the master stores that worker's actor ref inside its internal State uh the worker can also request work from the actor this usually happens once uh the first time it registers itself with the master as as as well as when it uh finishes work and there's also a work is done message that serves a similar purpose to uh work or request work and we'll see an example of that in a bit uh so the master the Constructor is pretty simple uh we have some notion of an abstract algorithm and the algorithm is basically a trait that defines a single method execute the execute method takes in uh two parameters one is the graph structure and one is uh some sort of input so maybe a node ID or a pair of nodes the result Handler is where all the results of the computation gets sent so once a so for example a master will send to a worker uh a single node ID the worker will compute the full breath for search across the entire graph it'll take that map send it to the result Handler and the result Handler can choose to do some sort of reduction on it or just dump that map to disc or serialize it or whatever and by default the most common use case for this for us was just uh literally just pretty printing out a map onto dis but there were some cases where uh I did do some sort of reduction you can also see the two uh important data structures workers is a map from a workers actor ref to uh any work it's doing so it's an option because that worker may be Idol or may be doing some sort of work and the work queue is basically just a cue of uh every any work that's left to be done so the first message that the master can receive is worker created uh it's pretty straightforward it'll log in it'll watch it what context. watch does is basically um that's it's sort subscribes the master to the worker's events so if that worker were to Die the master would get a terminated message and then the master can choose uh what to do with that message uh in this case he would take whatever work was uh it would take whatever work that act that actor was working on and then put it back in a queue and then consider that actor dead uh it will so once it receives the worker created that means this is the first time a worker has contacted the master so it will go ahead and bundle up that algorithm that was in the Constructor send it to the worker so the worker now knows what algorithm to compute and then it will notify the worker um it will call this notify workers function which basically all it does is if there is work left to be done inside a queue it will broadcast a message to all workers saying uh I have work if you're Idol talk to me and I'll give you work uh the second message worker request work uh is pretty simple it make sure that the worker has registered before if the work is empty it just sends back there's no work to be done and you can send an idol stage uh otherwise uh if you are Idol then IQ something off the work I put into the mapping that you are doing work and then I tell you uh the work that I want you to do and notice that we're not just doing the we're not just doing worker bang uh work to be done work we're using this worker. tell so I can do this result Handler thing so uh to the worker the sender is the result Handler and so it can automatically use the sender to uh send back whatever the result is uh the work is done message uh pretty straightforward if the worker is not has never registered before you log it otherwise make it idle uh at this point if if uh all work is finished meaning the queue is empty all work the all work send uh message has been received that's when the master begins broadcasting messages to tell everyone to shut themselves down uh this is the message that the ma that the master will get if a worker were to die for whatever reason uh basically what it does is log the error get the work that that worker was previously working on and then tell itself to distribute a work so that triggers that event where it just takes the encapsulated job encapsulated work and puts it back in a queue and then removes the worker from the map it doesn't try to do any sort of recovery for the worker if a worker goes down then it's down uh this event gets triggered when a remote jvm completely shuts down and that just triggers a function that goes through the entire worker map and sees which workers were on that machine and considers them dead and then recues to work uh this is Trivial you get work queue it up notify workers that just need work to be done and so if there are idle workers and uh they're just sitting there once you get new work you should wake them up they shouldn't be sitting idly by just doing nothing and wasting resources and all works sent just triggers a simple flag and then at the very end if you get some sort of message for whatever reason you just log error and you're done on the worker end uh it only ever really communicates with the master the master will give a due algorithm which is which will it will receive the first time it registers with the master uh it will get work to be done from the master after it asks for work uh they'll be work is ready when the master calls the notify workers function and then it can also get no work to be done when there's nothing left in the queue and that's when the actor Will idle uh the Constructor to the worker so each worker will get its own copy of the graph it's not sent to it by the by the master there's no reason for it to everyone's sitting on NFS so they can see the graph anyways and then it gets a reference to the master so that's how it knows to send a message to the master to trigger the registration uh you have this case object work complete that's a message it sends to itself after the work is done and I'll show you uh where exactly that's used in a bit uh you start out with no algorithm you just have have some you just know you hope in the future that you get an algorithm and if the system works fine you will before you get any sort of work uh before the worker actually starts that's the message that it sends to the master to register itself uh so the worker will be in two phases it will either be working or idle uh so inside the the default receive function that is set equal to the idle phase and I'll talk about uh how the idle phas the idle Behavior handles messages in a bit in a case of uh working Behavior if you're working and someone is telling you that work is ready or there's no work to be done that's meaningless to you you have work to do so there's no reason you should act on those messages uh this is uh so at some point when you're done Computing on your work you're going to send it the the worker is going to send a a message to itself that's the work complete message that we saw earlier in that case it sends the master it tells the master that's it's done with the work and then request more work and it switches into the idle stage uh in the idle stage this is the default this is how the worker start out of course uh they get the algorithm they they set the internal state to to that algorithm uh when workers ready uh if they get a work is ready message that's when they go to the master and ask for work and if there's uh when the M when the master gives them work it first makes sure that they have an algorithm there's the system runs fine there's no reason for it to have no algorithm uh the most I've never seen a case where it's where it's logged that error but it's good to log it just in case it does happen uh this is the interesting part when it gets something work to do it calls this do work function what do work does instead of actually Computing and blocking on that worker it will spawn off a future of that computation and at the very end of that future as part of the future computation it will send a message to the result Handler containing the result and then finally it will send itself a message the work complete message and that's what triggers uh the behavior that we saw here so do Works uh spawns a future that pipes a message to itself and then there's no work to be done then you remain in an idle stage this is the layer uh between the whole actor system and the outside world uh this is what clients will users will only ever use they will call saber. execute give it the abstract algorithm object and a work which is an itable of any and then potentially if they want to create their own result Handler actor which I've used before they can they can specify their own output file name or they can uh just go with the default saber result.txt uh a full example of an application looks like this and that's literally all the code there is uh minus the Imports so this is the simple degree algorithm I was talking about earlier where each each worker gets a single node ID and then it will find the degree of that node and then send it back so it extends abstract algorithm uh it overrides the execute method uh and then it simply uh inside the graph gets the node gets the node ID calls the degree method on it wraps it up in a result and then sends it back to the to the result Handler inside the main object uh you define the work that you want so there's a convenience function inside saber called from list. noes uh the user can choose to specify on the command line if they want to if they want to write to a different uh output file other than the default saber result.txt and so the general flow of the application looks a bit like this uh you have the application on top which in that uh last example would have been the degree application you have saber and you have mastering of result Handler so what happens is the application called saber. execute that will spawn uh the saber uh actor system saber will then uh spawn a master and a result Handler once that's all set up uh workers can begin spawning and registering themselves with the master and so they'll have a worker created message the master will say do this algorithm followed by work is ready message worker will request work Master will uh give it the work to do the worker will begin Computing as uh in the future as part of that future it will send a message to the result Handler called handle result the result Handler can choose to do a reduction or write it to file or whatever it will then send itself a work complete message and receiving that work complete message triggers itself to send a work is done message to the master and this can be done uh this is this process is done across all workers and at any point during this computation U more workers on different machines can be spawned and to go through exact same process you register with the master and the whole process continues and so you might have seen in a code that there was a lot of uh NES and this is a consequence of AA being currently not typed they're untyped endpoints receive is actors are basically a function from any to unit and so uh some future work that I was interested in are type channels which use Scala macros but I'm hesitant on using them at the moment because macros are in an experimental phase and I don't want to dedicate a huge chunk of my time to something that may be phased out so that would be interesting uh AA clustering became finalized in AA 2.2 and that implements a Dynamo like peer-to-peer clustering system where you have gossiping and a lot of this cool fa tolerance and that might be interesting and recently they U made the types safe developer console free for developers and that's basically a really fancy web UI that let you monitor all the messages being sent throughout your system uh the the statuses of your your various actors and that and there these all these amazing uh monitoring metrics that they have and it's worth checking out uh if we have time I believe okay we have plenty of time uh let's be a hopefully this will go well okay okay so I'm on a remote machine on my lab right now uh so does that help can people see my code just the lights oh okay yeah so uh yeah I have my build IBT file uh the graph Library I'm using is uh scholar graph and that's dependency uh I use scholar iio to read and dump to disk and then that is the actual saber dependency uh I will run the degree algorithm which uh outside of those various Imports does it just comp the degree of which node yeah do the work splitting like how does it make how do you acheve the speed up so the user defines it so in this case we're doing it we're just getting a list of all the nodes and then uh we give it to we do saber I execute uh all nodes and the saber will just take each it treats each element of the list as a single yeah and so if you wanted a a per node thing you would group inside here you would find all pairs or a certain subset of pairs and then send that to saber so saber just treats each element as the as as the payload so yeah uh besides the Imports that was literally all the code that was needed uh what I do is I have the S spt assembly plugin so I build it into a Fatar this saber laab assembly. one snapshot uh I have a configuration file that tells me let me see if I can make this go away okay I have a configuration file that tells me which graph to compute on uh the address of the master node and then I have a bunch of mappings from uh worker worker addresses to how many threads they should use uh there's also I also have a short script that basically parses this saber deployed. CFG file and anything that's not commented by a pound sign it will automatically deploy it's a python script that will deploy workers to those uh to those machines and so first I'm going to spawn uh a master so uh I believe the script was deploy Master yeah so just ases s spt run main uh gives the class path and then it give us the output file so currently there's no all my degrees

text file in here hopefully by the end we'll have that so I'm going to deploy master all right Master deployed uh I'm going to go into a different screen and then I have this uh deploy workers that's basically going to go through that saber deploy CFG file and spawn workers on each machine there's no Python 2.7 so we'll do that instead okay there's a bunch of logging being done uh let's see if I can find where okay so at that point this the graph is small enough that it's finished Computing but you can see that it's getting uh uh IT workers are registering from Ant-Man Captain America our our machines are named after Marvel superheroes but uh you can see now we have an all my degrees. txt file and if we look inside it it's just a mapping from node ID to the degree of that note and so we can I have that was just a trival example to make sure that the code terminated in time we've done uh things like uh the all pair shortest path page rank uh random walk distances things like that but yeah the goal was to get uh minimal overhead for spawning an application hopefully I've convinced you that that was minimal overhead and and yeah that's it for my presentation any questions so basically the model is that the graph is small enough to fit in Ram machine so the speed up is achieved by basically thinking of the algori which will start different no and do different work right so so that the model like your algorithms are expensive you want to but you know graphs fit uh you will like there work sharing essential like if you need parameters intermed steps from to the other you don't communicate yeah so this is only for the embarrassingly parallel algorithms like the like the all shortest P stuff any other questions okay cool thank you