Hey all,
maybe you've read it already on Twitter, I got picked for GSoC2k11 :D
Of course I'm very proud and want to thank everybody who helped me across the way. Especially my mentor: Edward J. Yoon.
If you don't know what my project actually was, have a look at Development of Shortest Path Finding Algorithm.
Basically I want to create a scalable solution, that takes a HUGE graph :) and and provides a fast solution of SSSP (single source shortest path) problems.
Like I already told in the Apache's Jira issue, I focus on a distributed Dijkstra algorithm that uses an adjacency list. So at first I'll focus on sparse graphs. The adjacency list is stored into a BigTable structure, in my case this will be HBase.
After I finished this, I will focus on a refinement of partitioning and if I have enough time I extend the Dijkstra with an A* heuristic.
These steps are basically my plan for this summer :)
Right now I started my first commit of Dijkstra's algorithm, it currently contains the old BSP version which I started with weeks ago to get familiar with Apache Hama. It is actually a working example, so feel free to check out the code and run its main method. This will start a local multithreaded BSP with some sample data of the german wikipedia page of Dijkstra's algorithm.
The class is located here: de.jungblut.hama.bsp.BSPDijkstra.
The repository can be found here: http://code.google.com/p/hama-shortest-paths/
Please star this project and the jira issue :D
I'll keep you updated on every change and thoughts I made.
Thank you very much: Edward, Apache Software Foundation and Google!
Showing posts with label GSoC. Show all posts
Showing posts with label GSoC. Show all posts
May 1, 2011
Apr 12, 2011
Apache Hama Partitioning
Hello there!
I was a bit busy today after I've read the Google Pregel paper. It described quite good "how to think like a vertex" and how they are partition data across their cluster.
I wanted to share this new knowledge with you and I've shortly rewritten my Graph Exploration algorithm. It can be found as the post below or just following this link.
Just a short summary of what we are going to change:
I really felt dumb today after I read the Pregel paper. They are just partitioning the data with a hashfunction. And I was like running several MapReduce jobs to determine the optimal distribution of an adjacency list. So scratch that, we just use a simple HashFunction that is none other than returning the id of the vertex. If you are having for example Vertex IDs' from 0-100 and you have to distribute it over 5 machines, just loop to 100 and modulo it with the 5 machines. Life can be so simple. sic!
How to distribute?
First off we share the common FileSystem across the servers. The idea is that every server get's its own SequenceFile where the data is stored in. Let's rush into the code that parses my AdjacencyList textfile into the SequenceFiles for the servers:
Basically we are creating a SequenceFile for each Groom and writing the data with the modulo function into the SequenceFiles. Note that the names of the SequenceFiles are related to the name of the peer. That is because we can simply let each peer find its partition.
How to pass messages between vertices using partitioning?
If we know what kind of partition we used, this is very simple. Look at the layered send method.
The only requirement is, that the indices of peer.getAllPeerNames() are the same we used in the partitioning phase, otherwise it will result in strange behaviour.
With the help of these two methods we are now able to make the main algorithm use less master-slave communication and therefore use a more collective communication.
But keep in mind that we have to use the master-slave communication to keep track of the main loop. The problem behind it is that if a slave breaks its calculation because there can't be more updates made in a superstep, the other peers will deadlock in the next sync phase because one slave already has finished.
So we have to sync the updates with a master task and broadcast whether we can break the main loop altogether (if no task can update anymore) or we need another superstep.
This is actually a bit hacky. In the Pregel paper it is called "voteToHalt". Have a look at my implementation of the same:
A bit more human readable it says: Every task sends how many vertices are stored in the activeList and a master decides whether to break or not. This decision is broadcasted again and a false return value will break the outer running loop.
Keep in mind that this is not very very very optimal.
Output
Now we have multiple files in HDFS that can now be read or merged to get the classified components of the graph.
This snippet is quite generic if you want to use it in your application: feel free to copy and paste ;)
As you can see, it just fetches the parent directory of the outputs' and loops over the files while writing the content to STDOUT.
Gladly this helped a guy on Stackoverflow today. And helped me to get my 500 reputation :)
Finally it gaves the same output as the other algorithms:
The new class is called: de.jungblut.hama.graph.BSPGraphExplorationPartitioned.
Just execute the main method, it will run a local multithreaded BSP on your machine.
This class is located in my Summer of Code repository under: http://code.google.com/p/hama-shortest-paths/
Star this project or vote for my GSoC task.
The next week I'll go for a comparision between these three implementations: partitioned-BSP, BSP and MapReduce.
But I guess before it I focus at PageRank. Some guys of Apache Nutch wanted a distributed PageRank in BSP, I'll go for a working example next time.
The Pregel paper said something about 15 lines of code. Looks like a short task ;)
[UPDATE]
If you are interested in a more faster way to partition, check out the new blog post here:
http://codingwiththomas.blogspot.com/2011/08/apache-hama-partitioning-improved.html
bye!
I was a bit busy today after I've read the Google Pregel paper. It described quite good "how to think like a vertex" and how they are partition data across their cluster.
I wanted to share this new knowledge with you and I've shortly rewritten my Graph Exploration algorithm. It can be found as the post below or just following this link.
Just a short summary of what we are going to change:
- Partition the vertices across the servers.
- Messages that are related to a specific vertex will just go to the server the vertex was partitioned to.
- The master task just keeps track of when the task is finished (no updates were made).
- Each task has its own output.
I really felt dumb today after I read the Pregel paper. They are just partitioning the data with a hashfunction. And I was like running several MapReduce jobs to determine the optimal distribution of an adjacency list. So scratch that, we just use a simple HashFunction that is none other than returning the id of the vertex. If you are having for example Vertex IDs' from 0-100 and you have to distribute it over 5 machines, just loop to 100 and modulo it with the 5 machines. Life can be so simple. sic!
How to distribute?
First off we share the common FileSystem across the servers. The idea is that every server get's its own SequenceFile where the data is stored in. Let's rush into the code that parses my AdjacencyList textfile into the SequenceFiles for the servers:
public static List<Path> partitionAdjacencyTextFile(int sizeOfCluster,
Map<String, String> groomNames, Path fileToPartition,
Configuration conf) throws IOException {
// setup the paths where the grooms can find their input
List<Path> partPaths = new ArrayList<Path>(sizeOfCluster);
List<SequenceFile.Writer> writers = new ArrayList<SequenceFile.Writer>(
sizeOfCluster);
FileSystem fs = FileSystem.get(conf);
for (Entry<String, String> entry : groomNames.entrySet()) {
partPaths.add(new Path(fileToPartition.getParent().toString()
+ Path.SEPARATOR + "parted" + Path.SEPARATOR
+ entry.getValue()));
}
// create a seq writer for that
for (Path p : partPaths) {
fs.delete(p, true);
writers.add(SequenceFile.createWriter(fs, conf, p,
IntWritable.class, IntWritable.class));
}
// parse our file
FSDataInputStream data = fs.open(fileToPartition);
BufferedReader br = new BufferedReader(new InputStreamReader(data));
String line = null;
while ((line = br.readLine()) != null) {
String[] arr = line.split("\t");
int vId = Integer.parseInt(arr[0]);
LinkedList<Integer> list = new LinkedList<Integer>();
for (int i = 0; i < arr.length; i++) {
list.add(Integer.parseInt(arr[i]));
}
int mod = vId % sizeOfCluster;
System.out.println(vId + " goes to " + partPaths.get(mod));
for (int value : list) {
writers.get(mod).append(new IntWritable(vId),
new IntWritable(value));
}
}
data.close();
for (SequenceFile.Writer w : writers)
w.close();
return partPaths;
}
Basically we are creating a SequenceFile for each Groom and writing the data with the modulo function into the SequenceFiles. Note that the names of the SequenceFiles are related to the name of the peer. That is because we can simply let each peer find its partition.
How to pass messages between vertices using partitioning?
If we know what kind of partition we used, this is very simple. Look at the layered send method.
private void send(BSPPeerProtocol peer, BSPMessage msg) throws IOException {
int mod = ((Integer) msg.getTag()) % peer.getAllPeerNames().length;
peer.send(peer.getAllPeerNames()[mod], msg);
}
The only requirement is, that the indices of peer.getAllPeerNames() are the same we used in the partitioning phase, otherwise it will result in strange behaviour.
With the help of these two methods we are now able to make the main algorithm use less master-slave communication and therefore use a more collective communication.
But keep in mind that we have to use the master-slave communication to keep track of the main loop. The problem behind it is that if a slave breaks its calculation because there can't be more updates made in a superstep, the other peers will deadlock in the next sync phase because one slave already has finished.
So we have to sync the updates with a master task and broadcast whether we can break the main loop altogether (if no task can update anymore) or we need another superstep.
This is actually a bit hacky. In the Pregel paper it is called "voteToHalt". Have a look at my implementation of the same:
private boolean voteForHalt(BSPPeerProtocol peer, String master)
throws IOException, KeeperException, InterruptedException {
peer.send(master, new IntegerMessage("", activeList.size()));
peer.sync();
if (peer.getPeerName().equals(master)) {
boolean halt = true;
IntegerMessage message;
while ((message = (IntegerMessage) peer.getCurrentMessage()) != null) {
message = (IntegerMessage) peer.getCurrentMessage();
if (message.getData() != 0) {
halt = false;
break;
}
}
peer.clear();
for (String name : peer.getAllPeerNames()) {
peer.send(name, new BooleanMessage("", halt));
}
}
peer.sync();
BooleanMessage message = (BooleanMessage) peer.getCurrentMessage();
if (message.getData() == true) {
return false;
} else {
return true;
}
}
A bit more human readable it says: Every task sends how many vertices are stored in the activeList and a master decides whether to break or not. This decision is broadcasted again and a false return value will break the outer running loop.
Keep in mind that this is not very very very optimal.
Output
Now we have multiple files in HDFS that can now be read or merged to get the classified components of the graph.
This snippet is quite generic if you want to use it in your application: feel free to copy and paste ;)
private static void printOutput(FileSystem fs, Configuration conf)
throws IOException {
System.out.println("-------------------- RESULTS --------------------");
FileStatus[] stati = fs.listStatus(new Path(conf.get("out.path")
+ Path.SEPARATOR + "temp"));
for (FileStatus status : stati) {
Path path = status.getPath();
SequenceFile.Reader reader = new SequenceFile.Reader(fs, path, conf);
IntWritable key = new IntWritable();
IntWritable value = new IntWritable();
while (reader.next(key, value)) {
System.out.println(key.get() + " | " + value.get());
}
reader.close();
}
}
As you can see, it just fetches the parent directory of the outputs' and loops over the files while writing the content to STDOUT.
Gladly this helped a guy on Stackoverflow today. And helped me to get my 500 reputation :)
Finally it gaves the same output as the other algorithms:
-------------------- RESULTS -------------------- 0 | 0 1 | 1 2 | 2 3 | 2 4 | 1 5 | 2 6 | 2 7 | 1 8 | 2 9 | 0
The new class is called: de.jungblut.hama.graph.BSPGraphExplorationPartitioned.
Just execute the main method, it will run a local multithreaded BSP on your machine.
This class is located in my Summer of Code repository under: http://code.google.com/p/hama-shortest-paths/
Star this project or vote for my GSoC task.
The next week I'll go for a comparision between these three implementations: partitioned-BSP, BSP and MapReduce.
But I guess before it I focus at PageRank. Some guys of Apache Nutch wanted a distributed PageRank in BSP, I'll go for a working example next time.
The Pregel paper said something about 15 lines of code. Looks like a short task ;)
[UPDATE]
If you are interested in a more faster way to partition, check out the new blog post here:
http://codingwiththomas.blogspot.com/2011/08/apache-hama-partitioning-improved.html
bye!
Apr 8, 2011
Back to Blogsphere and how BSP works
Hey guys,
I'm back! For those of you that do not remember me: I did some cool Robots for Google Wave.
Maybe you remember the times where no export function was in Google Wave, I targeted it with my bot called "PorteXy". It was a Java portation of the existing Google example called "Exporty". From the count in the AppEngine database table I know a lot of you used it and I wanted to thank you all for your support!
Now we all now that Google Wave passed by and gets an Apache Incubator project, I have to focus on other themes that could potentially be interesting for you. Since the last time, a lot has happened.
I am no pupil anymore, I started working at testberichte.de, recently got my OCP in Java and in the mean time I am studying Computer Sciences at HWR Berlin.
You must be thinking: "Oh he must be really busy" - in fact I am not. School kinda sucks, because it is too boring and I have to do something useful. So you've probably heard that Google is offering a new Summer of Code this year: GSoC2011. And I decided to apply to it, my first thought was that I am going to apply for the Apache Software Foundation.
Mainly because I worked a lot with Hadoop and Distributed Systems in general and loved to use the Apache2 server running my PHP applications or my Tomcat running my Java applications. But Hadoop wasn't a project eligible for GSoC (Why? I know it has a really large codebase, but there are so many cool task that can be done by students too!), so I had to look for another project. Then I've seen Mahout, I used Mahout for k-means clustering recently and looked at the task offering: Support more data import mechanisms. Are they serious? Just importing raw data into SequenceFiles? Seriously, you can't use 12(!) weeks for that task coding 40h/week. This would be a task for a weekend, if at all. So I was looking for something more cool and complex, but this should be in Java and somehow related to distributed systems.
So I came across Apache Hama and this task is smiling at me: Development of Shortest Path Finding Algorithm. BÄM! This was the task I've searched for, I love distributed Computing and I love Graphs.
You probably know that Hama uses BSP (Bulk synchronous parallel) for its computing.
This is actually an abstraction to MapReduce. Have a look at this picture on wikipedia:
If you translate MapReduce into BSP, then your map-phase will be the local computation-phase. After that you are going to merge the map output and sort it. That would be the communication phase. Now comes the Barrier Synchronisation: You know that no reducer can run if not all map task completed. So this step is a bit merged with the communication phase, but after that it is entering a new local computation phase: the reduce-phase.
So you see, BSP is not the same like MapReduce, but you can actually describe MapReduce with BSP.
What is the advantage of using BSP instead of MapReduce in Graph Processing:
Those of you who implemented graph algorithms with MapReduce probably know what a pain it is to pass messages through HDFS and process them in several chained jobs. I believe it was this paper here, what described very well how this works: http://www.umiacs.umd.edu/~jimmylin/publications/Lin_Schatz_MLG2010.pdf
Even if you don't know it how to implement them (maybe I'll go for a tutorial later on) believe me: this is not optimal!
Why should we pass messages through DFS? Well, just communicate the messages!
And that is the real advantage of BSP, you having less sync (in MapReduce you have 3 major sync steps in a job: the beginning, the reducer and the end. The end, because you can't run a follow-up job before the last one finishes, and therefore the beginning too) and a more "real-time" messaging than MapReduce.
BSP is actually used at Google too: http://googleresearch.blogspot.com/2009/06/large-scale-graph-computing-at-google.html, you can google for the Pregel paper for further reading.
In my opinion Apache Hama has a great perspective, because there is no other OpenSource alternative for BSP processing and the best thing of all: I'll be part of it.
Back to the shortest path finding. As you can see, I already started to implement a basic Dijkstra and made some thoughts on how this is going to work within a large cluster, mainly in terms of scalability.
If you are interested in shortest paths in large graphs you should definitely bookmark this issue or at least vote it up.
The application deadline of GSoC is in approx. 10 hours and I hope ASF will pick me :-)
This was a really large first posting and I want to thank especially Edward J. Yoon, he is my mentor at Apache Hama and told me that I should blog again. This is part of his "young open source stars"-support ;) Visit his blog or follow him on twitter, he is a real genius!
Obviously you can also follow me on twitter, look at the sidebar on the right side.
Greetings from Germany!
I'm back! For those of you that do not remember me: I did some cool Robots for Google Wave.
Maybe you remember the times where no export function was in Google Wave, I targeted it with my bot called "PorteXy". It was a Java portation of the existing Google example called "Exporty". From the count in the AppEngine database table I know a lot of you used it and I wanted to thank you all for your support!
Now we all now that Google Wave passed by and gets an Apache Incubator project, I have to focus on other themes that could potentially be interesting for you. Since the last time, a lot has happened.
I am no pupil anymore, I started working at testberichte.de, recently got my OCP in Java and in the mean time I am studying Computer Sciences at HWR Berlin.
You must be thinking: "Oh he must be really busy" - in fact I am not. School kinda sucks, because it is too boring and I have to do something useful. So you've probably heard that Google is offering a new Summer of Code this year: GSoC2011. And I decided to apply to it, my first thought was that I am going to apply for the Apache Software Foundation.
Mainly because I worked a lot with Hadoop and Distributed Systems in general and loved to use the Apache2 server running my PHP applications or my Tomcat running my Java applications. But Hadoop wasn't a project eligible for GSoC (Why? I know it has a really large codebase, but there are so many cool task that can be done by students too!), so I had to look for another project. Then I've seen Mahout, I used Mahout for k-means clustering recently and looked at the task offering: Support more data import mechanisms. Are they serious? Just importing raw data into SequenceFiles? Seriously, you can't use 12(!) weeks for that task coding 40h/week. This would be a task for a weekend, if at all. So I was looking for something more cool and complex, but this should be in Java and somehow related to distributed systems.
So I came across Apache Hama and this task is smiling at me: Development of Shortest Path Finding Algorithm. BÄM! This was the task I've searched for, I love distributed Computing and I love Graphs.
You probably know that Hama uses BSP (Bulk synchronous parallel) for its computing.
This is actually an abstraction to MapReduce. Have a look at this picture on wikipedia:
If you translate MapReduce into BSP, then your map-phase will be the local computation-phase. After that you are going to merge the map output and sort it. That would be the communication phase. Now comes the Barrier Synchronisation: You know that no reducer can run if not all map task completed. So this step is a bit merged with the communication phase, but after that it is entering a new local computation phase: the reduce-phase.
So you see, BSP is not the same like MapReduce, but you can actually describe MapReduce with BSP.
What is the advantage of using BSP instead of MapReduce in Graph Processing:
Those of you who implemented graph algorithms with MapReduce probably know what a pain it is to pass messages through HDFS and process them in several chained jobs. I believe it was this paper here, what described very well how this works: http://www.umiacs.umd.edu/~jimmylin/publications/Lin_Schatz_MLG2010.pdf
Even if you don't know it how to implement them (maybe I'll go for a tutorial later on) believe me: this is not optimal!
Why should we pass messages through DFS? Well, just communicate the messages!
And that is the real advantage of BSP, you having less sync (in MapReduce you have 3 major sync steps in a job: the beginning, the reducer and the end. The end, because you can't run a follow-up job before the last one finishes, and therefore the beginning too) and a more "real-time" messaging than MapReduce.
BSP is actually used at Google too: http://googleresearch.blogspot.com/2009/06/large-scale-graph-computing-at-google.html, you can google for the Pregel paper for further reading.
In my opinion Apache Hama has a great perspective, because there is no other OpenSource alternative for BSP processing and the best thing of all: I'll be part of it.
Back to the shortest path finding. As you can see, I already started to implement a basic Dijkstra and made some thoughts on how this is going to work within a large cluster, mainly in terms of scalability.
If you are interested in shortest paths in large graphs you should definitely bookmark this issue or at least vote it up.
The application deadline of GSoC is in approx. 10 hours and I hope ASF will pick me :-)
This was a really large first posting and I want to thank especially Edward J. Yoon, he is my mentor at Apache Hama and told me that I should blog again. This is part of his "young open source stars"-support ;) Visit his blog or follow him on twitter, he is a real genius!
Obviously you can also follow me on twitter, look at the sidebar on the right side.
Greetings from Germany!
Subscribe to:
Posts (Atom)