Quick guide: http://wiki.apache.org/hama/GSoC2011

Would you take into consideration sparse graphs or dense graphs?

Weighted or unweighted? Directed or undirected? Or all the possible combinations of these two?

I think it should at least be scalable, no matter if sparse or dense.
Weight or unweight is just a matter of what's defined by the user, the algorithm should IMO deal with negative weights, zero weights (or unweighted) and of course positive weights.
12 weeks are enough time for all permutations

So I've implemented a few graph algorithms using MapReduce and can't wait for this task here using BSP. I'm gonna apply to it.

1+ overall

+1 ! 12 weeks are enough time for all, let the user choose how he want to start.

Note that you can freely design your own scenario. We'll help you!

For example, you can just focus on to show the advantages (e.g., sequential programming, scalable, performance, ..., etc) of using Hama BSP to graph computing.

I read the papers about microsoft trinity Alois posted on Twitter. It was quite interesting.
Maybe we should need something equal to the graph database, HBase would be an option. But since it is no more a requirement for Hama, I tend to write a SequenceFile Database.
Storing byte offsets to a special lookup key or something..

Had a bit time and worked through Hama and BSP!
Never the less I have another question:
What is the advantage to use a "master groom" like in the PiEstimator example? I mean this could turn into a large bottleneck. BSP uses the communication phase to communicate between the nodes and not the nodes to a master node. Or did I missed something?

Thanks!

Hey guys,

About using HBase as graph database, the tinkerpop team did a great job implementing blueprints on top of HBase:
https://github.com/dgreco/graphbase

Maybe this could help ? or we could collaborate with them ?

Cheers,

Hey Alois,

graphbase looks quite useful. Did you already try using it?

Hello Thomas,

Not at all, was released yesterday !

I'm actually in touch with the developer (David Greco), for the next steps he want to focus on creating MapReduce algorithm on top of graphbase...

But I want to dive into the code this week-end and see what could be usefull for Hama, this could be exactly what we need for GraphDB integration (and I think integration with GraphDBs is important)... But we could perhaps use Blueprints instead of HBase directly in that way Hama could be used on other GraphDB too

I'm perhaps too optimist (and don't really know how graphbase work) but that's definitely sounds good.

What your thoughts on that ?

Cheers,

Just want to add that Blueprints can't be used the way I was thinking, no 'data locality awareness' (didn't find better way to express), it's just a simple layer to access a graphdb.

Even M/R algorithms David will develop gonna be tied to Hadoop/HBase, so we are...

But I'm still convinced about the usefulness of BSP on HBase as GraphDB.

Cheers,

Hey guys,
just to pointed out that I'm not part of the tinkerpop team. I decided to use their very nice apis to drive my development effort. I think I found a way for laying out a graph on top of an HBase table. I tried to make all the graph accesses only based on accessing a single HBase row, since updating an hbase row is atomic in this way I can make accessing/modifying graph vertexes, nodes (almost) atomic.
Than, given the fact that Hbase supports M/R, I'd like to implement some M/R graph algorithm to see how it goes.
Please, feel free to contact me for questions/suggestions.
Regards,
David

Hehe, I should've looked at the age of the files :o)

The advantage of using HBase is the following:
HBase is built on top of HDFS, so it is controlling the distribution of the data.
BUT this could be a large disadvantage too, in the case if a groom is not running on a server where the data is actually stored.
In MapReduce this is called a non-local task, so you have to copy the data to the local datanode.

Maybe Edward can tell us why he decided to get rid of HBase and MapReduce at all.

Using a GraphDB and an interface like Blueprints is like using a MySQL Database with JDBC inside a distributed environment. It is possible, but IMHO it is not optimal.

My scenario would be:

• you have vertices represented as edges inside of a SequenceFile in this style: inVertex;outVertex;weight; and so on... (quite similar to graphbase)
  • the user could then simply setup a MapReduce job that connects with Blueprints and write it into the SequenceFile.
• we build an index on top of this, let this simply be a hashmap with a vertex and its byte offset in the sequence file
  Graphbase could make this work for us.

• then we focus on some basic Dijkstra, so the weights are positive, no matter if the graph is directed or not.
  • we see how we can distribute this over many machines
  • my first approach would be a master server that holds a shortest-path-matrix and let the slaves calculate the distance between the vertices and after that communicate with the master, it simply sums them up and updates the matrix.
• for negative weights and very large datasets we could implement a A* heuristic

This is not really optimal, I know that. But this would be my first approach on this topic.
So what are your thoughts on that?

I think, the best-fit structure for HBase table, is a adjacency matrix. Google also stores "web link graph" to BigTable using an adjacency matrix. If you want to use HBase, you can simply do it.

I looked at the graph-hbase, but it's just a graph-friendly API layer on top of HBase. It'll make you feel complex.

>> Maybe Edward can tell us why he decided to get rid of HBase and MapReduce at all.

Keeping in mind that it doesn't mean that the HBase isn't fit to store the matrix/graph.

Okay so scratch that SequenceFile stuff.
Did not thought about an adjacency matrix in HBase.

Thanks Edward

>>BUT this could be a large disadvantage too, in the case if a groom is not running on a server where the data is actually stored.
>> In MapReduce this is called a non-local task, so you have to copy the data to the local datanode

Exactly what I was thinking when I was speaking about problem of data locality.

>> Using a GraphDB and an interface like Blueprints is like using a MySQL Database with JDBC inside a distributed environment. It is possible, but IMHO it is not optimal.

+1 Same conclusion here. I would say it's even an horrible abstraction inversion.

>>>>I looked at the graph-hbase, but it's just a graph-friendly API layer on top of HBase. It'll make you feel complex.

So ok, no added value at all. Will break the flexibility of storing the adjacency matrix the way you want.

To concluded it's seems more important to know which data structure to use, more than where to store it.

When you sure which structure to use (adjacency matrix i.e.) you can then choose the best system to store/access it (SequenceFile/HBase/...) and change it if necessary without impacting the algorithm.

Thanks Edward and good luck Thomas !

For A* and Dijkstra it is crucial if the data is not local, especially if you keep the distances in an HBase.
We'll see, I think this gonna work well.

I'm mentoring this project.

• name: Edward J. Yoon
• link_id: edwardyoon

> Never the less I have another question:
> What is the advantage to use a "master groom" like in the PiEstimator example?
> I mean this could turn into a large bottleneck. BSP uses the communication phase to
> communicate between the nodes and not the nodes to a master node.
> Or did I missed something?

Hi,

http://wiki.apache.org/hama/PiEstimator

In PiEstimator example, master task is one BSP processor of nodes, not a master server of cluster.

Yes. that's right, I meant the code that only executes on the "master".

Do you require a full design of the algorithm in the application?

Since current version of Hama doesn't provide an I/O API and data partitioner, it might be hard to implement all.

As I mentioned above, please focus on understanding of the BSP model, and to show the advantages (e.g., sequential programming, scalable, performance, ..., etc) of using Hama BSP to graph computing.

Here's my example scenario:

1. You can create your own adjacency matrix of graph G on HBase.
2. Then, you can assign range 1-10 to task-1, 11-20 to task-2, ...
3. In bsp() function, you can scan the assigned range of HBase table.

If you know how many grooms are available and how large your dataset is, this will be the least problem.

I don't think that you can spend 12 weeks on just showing advantages of BSP.

Actually, graphbase is using an hbase table as an adjacency matrix. I put a row per vertex and all the edges ids are just qualifier for the column family where the cell value is the in-vertex id. In the same row I'm also putting the vertex property and the edges properties. I'm just using the convention to store the edge properties on the row corresponding to the out vertex. I tried to put as much as possible everything in one table per graph, and also being able to do update the graph information updating maximum two rows. Sorry, I didn't have time to put too much information on the github site, this is something I'd like to sort out soon.

Thomas,

http://www.cs.rice.edu/~vs3/comp422/lecture-notes/comp422-lec24-s08-v2.pdf

This slides will be helpful for you.

Wow this is awesome, thank you!

So I made up some algorithm's with the use of the slides. (not really optimal cause it will require a sync step per current start vertex)
I'll write a bit more later.

For now I have set up a google code repository to store the things needed.
It is quite large because I already generated some test datasets with 2 bn vertices based on cities in the world.
I wrote a mapper that will produce a random adjacency list that will be put into hbase in the reduce step.

The table layout will be just a simple adjacency list like:
vertex : rowID of the out-vertex and weight : rowID of the next out-vertex and its weight : etc.
So quite similar to David's approach, but much simpler.

Like the slides said, it should be quite difficult to distribute it even over the clusters. I have a cool partitioning idea for that
So more about it later.

Here is the code site: (full repository is about 400mb large)
https://code.google.com/p/hama-shortest-paths/

Including a sequential A*, Dijkstra, some hardcoded (don't blame me!) example graphs and a benchmark for it.

Good start and you're doing a good job.

Great I'm eager to look at this !

Thanks

Yesterday I setup the HBase and I decided to go a different way in the table layout.
The adjacency list will just contain the row numbers and the weights.
And you have a "lookup-table" that maps the vertex name, in my case the city name to the rownumber.

adjacency_list => rowId : rowId of the out-vertex and its weight
vertex_lookup => vertex : rowId

In the calculation of the shortest paths you don't need the name, so this won't blow up the table too much.
But if you are really looking for the shortest way from London to Paris you have to translate the city name to the rowid in order to start the shortest paths.
I guess this is a much better approach.

The first naive implementation of dijkstra would be:

You partition the row's to available grooms as boundaries (such as LIMIT's in a database, e.G from row 0 to row 10k). I know that if you have an adjecency list, the distribution is not perfectly balanced, I will provide a more advanced algorithm later.
Then you pass the grooms a message which tells them what is the current start-vertex.
Every groom is now searching for the lowest-weighted adjacent vertices and passes these messages to a "master-groom" like in the PI Estimator.
After that you are syncing, the master will detect the globally clostest vertex and broadcasts this as the new start-vertex to the grooms. This will update distances in an distance vector D (source: the slides).
Of course you have to actually repartition the datasets, because you can strike-through the row of the last start vertex, which will reduce the "height" of the list.
Then this steps will go over and over again until you have reached the breaking condition.

Note that this is just a few thoughts, there are some questions left:
How to store the already visited vertices? Or in the slides this is called the distance vector D.
How to store the direct ancestor of a vertex to reconstruct the shortest paths?
Is this optimal in the case of a sync step for every vertex in the graph?
etc.

So I will have a few things left for the actual summer of code
Thank you guys for your support.

A very similar program (based on openMP) is here : http://heather.cs.ucdavis.edu/~matloff/OpenMP/Examples/NM/Dijkstra.c

Thanks, have not seen OpenMP for a while I'm compiling this and step through the code tomorrow.
For now my test set is finished, I have arround 2 million vertices and 200 million edges in my HBase.
The next step is the partitioning idea. Write more about it later

I think, global variables also can be handled by HBase.

I guess storing the distance vector D in a HTable is a good idea, this table will also handle the direct ancestor.

For the first naive partitioning idea it is necessary to call this method:
http://hbase.apache.org/apidocs/org/apache/hadoop/hbase/client/HTable.html#getEndKeys%28%29

This will return your maximum row key, in this case these are positive ID's, so the range is going from zero to the max (return of the method).
Afterwards you are distributing the ranges across the available grooms.

For an adjacency matrix this is going to be a quite good partitioning: you always have the same row length.
In my case this is a list, so you have to watch out for edge hotspots (in my example there is a fixed number of edges, so this won't be critical at all).

The longer running partioning:
My advanced partitioning idea is that you'll setup a HTable scanner job, and you pass the number of grooms to the job via configuration.
Now the job (Map-Phase) determines how many row bytes for a specific key chunk (say 10k vertices, or at least based on the number of grooms) are currently inside this chunk.
So you'll be able in the reduce phase to determine which chunk has a hotspot, or if it is already evenly distributed. In the case of evenly distribution you know that you can use the naive partitioning.
But if it is not even, you'll have to chunk the hotspot(s) itself (say divide by 2). Write out the "chunked-hotspot" boundaries (or put it into a conf) and let the algorithm run again until it is better distributed.

So this is going to take a while since the real-partitioning is NP-complete. I read about Kernighan–Lin algorithm which is a heuristic:
http://en.wikipedia.org/wiki/Kernighan%E2%80%93Lin_algorithm
This is actually an edge-weight partitioning, we should replace the edge weight with the size of the row containing this edge.

Maybe you have a better (or at least faster idea?)

The next problem is the repartitioning:
So if you are using Dijkstra you can simply remove the already visited lowest-weighted vertex. So this row is going to be overhead for the grooms in the next step. If you are unlucky, every adjacency row in Groom1 will be dismissed and Groom2 has to do all the work on its own.
So to avoid this, you have to keep track of the boundaries that are passed to the grooms before a new step.

A big advantage of Google Pregel is the caching (didn't actually read the paper, but I read it somewhere else) so every "Pregel-Groom" has a real big amount of memory to cache the chunk he is always working on.
So we should not repartition too much, but this is currently just a thought so skip that caching stuff.

This is a cool page too:
http://www.eecs.berkeley.edu/~demmel/cs267/lecture18/lecture18.html#link_4.2

Checked in my first version of the BSP-Dijkstra. Working well.
This is not containing HBase, is just a "translation" of the sequential algorithm. Going to extend this with the partioning idea.

The containing Hama jar is the new 3.0 build that contains my local runner, so everbody just can run the main method in the dijkstra and see how the BSP works
I guess, I should now wait for the actual coding period

Naive partitioning is completed, currently working on the repartitioning.

Wow!
I'll look at it more deeply this week but, looks pretty good.

Great, I commit the changes this evening.
Note for me:
Profiling told me that finding the lowest vertex in the distance vector d will take much time. (in my current example cause of a full vertex scan in memory, in the real example this requires a full table scan in HTable)
Maybe storing a master local vertex id for the current minimum will reduce the overhead...

Is this gonna be just for the tropical semiring like Dijkstra's algorithm or other typical shortest path algorithms or is it gonna support a broader semiring framework?

For what I mean, google: mohri semiring framework shortest path

First of all, I focus on graphs and not on semirings. what does not mean, that we could not extend it by some time.
So first it will just deal with single-source SP and positive weights.

So I found that that bsp dijkstra won't deal with not connected graphs properly, I fixed that with the good old mapreduce vertex update counter
So the mainloop is now:
while(updated)
// do stuff
and not
while(we not visited all vertices)
// do stuff

If I would be evil, we could provide a real simple version of dijkstra in the next hama version with the wikipedia example I used.
So you would have a scalable and a teaching example.

for partitioning:
http://glaros.dtc.umn.edu/gkhome/metis/parmetis/overview

 There is one thing, I think, the proposal has a lack of detailed description of parallelization strategy.

Thank you Edward for your feedback, I'm going to describe it a bit clearer.
But I want to outline that this task strongly depends on the design of the input system. -> https://issues.apache.org/jira/browse/HAMA-258

What are the things we currently have:

• HBase table as input
  • this will be partitioned (key mod sizeOfCluster?)
  • every vertex is reachable with it's ID, the groom where the data actually is stored can be back-translated from the ID of a vertex.
• Through the partitioning phase we receive a subgraph on each groom
• Main algorithm works like this
  • We are broadcasting a start-vertex ID to the grooms
  • Now every groom is calculating the distances from the start-vertex to each adjacent vertex
  • Each groom holds a local minimum and sends this to a master task.
  • The master will review these minimas and broadcasts the new start-vertex (the global minimum) to the grooms.
  • Repeat until we can not update anymore.
• Output of each groom with its own vertex -> ancestor to reconstruct the shortest paths. Maybe we can store the actual weight, too.

Maybe we need to merge these. If you have a better idea of how we can deal with subgraphs more efficiently, you're welcome to give me a hint

Very nice!

Beside Metis/Parmetis there exists another partitioning tool named Scotch – http://www.labri.fr/perso/pelegrin/scotch/ – this is amongst others used in SHARC algorithm (e.g. i11www.iti.uni-karlsruhe.de/extra/publications/d-tdsr-09.pdf). This is a algorithm for shortest path (via Dijkstra), the incredible speed is achieved by massive preprocessing. The research of very fast dijkstra implementation seemed to begun for the 9th DIMACS implementation challenge (http://www.dis.uniroma1.it/~challenge9/). Hopefully the combination of both can be very interesting.

On there web-site of DIMACS the 10th challenge is for graph partitoning http://www.cc.gatech.edu/dimacs10/index.shtml ...

While searching for the address of SCOTCH i found this link https://gforge.inria.fr/forum/message.php?msg_id=105188&group_id=248 - there are some additional links at the end of the thread.

Andreas

Wow cool Thank you. If I have enough time I'll take a look to refine the partitioning with the links you provided.
Thank you very much for your help and the links!

Little update what I did the last two hours:

• Rewritten the GSoCDijkstra class in terms of fitting into HBase:
  • Did some fancy polymorphism to be able to test locally and use hbase on a cluster
    • added few interfaces and their implementations
  • remove the vertex class and just deal with the htable IDs to fit into the table layout

TODO:

• partitioning stuff (quite urgent)
• concrete hbase implementation
• vertexSet must be put into an abstraction layer in case of outsourcing into htable
  • currently this is just a hashmap representation of the original htable of the adjacency list
• Finish Output: make a more sophisticated method like every path from source with its cost
• afterwards do a full profiling of cpu and memory

So these things are checked in and running well so far.

I decided to put a few requirements on this project:

• Hbase 0.90.1
• Hadoop 20.2-append
• >Hama 0.2.0

I've finished the task using the SingleSourceShortest Path algorithm described in the Pregel paper.
We have now a working example that can be submitted.

Obviously the coding period hadn't started yet, so I will focus on my Dijkstra

Please checkout the new example and tell me what you think ;D

Here's the graph to validate that this works
http://de.wikipedia.org/w/index.php?title=Datei:DijkstraStep09.svg

Looks great

Thx.

I've added:

• a opportunity that the user can submit their own adjacency list as sequencefile
• the user can now choose from which vertex he want to start
• manipulate the output path

And I've added this to the example driver.
So this should be rdy, I'm submitting this to Hudson.

Hudson is hanging in a never ending timeout loop...

LOL, I've killed that job manually.

It is important to test locally before submitting patch.

I bet this wasn't my fault, I've done nothing with the tests. Of course I ran all the unit tests locally!

attach my patch.

Scheduling.

-1 overall. Here are the results of testing the latest attachment
http://issues.apache.org/jira/secure/attachment/12478502/eddie.patch
against trunk revision 1100148.

@author +1. The patch does not contain any @author tags.

tests included -1. The patch doesn't appear to include any new or modified tests.
Please justify why no tests are needed for this patch.

core tests +1. The patch passed core unit tests.

Changes : http://builds.apache.org/hudson/job/Hama-Patch/333/changes/
Console output: http://builds.apache.org/hudson/job/Hama-Patch/333/console

This message is automatically generated.

magic. You haven't changed anything in the patch.

See https://builds.apache.org/hudson/job/Hama-Patch/332/consoleFull and https://builds.apache.org/hudson/job/Hama-Patch/333/consoleFull

Your patch is always strange. How to generate patch?

edward@slave:~/workspace/hama-trunk$ patch -p0 < HAMA-359_v5.patch 
(Stripping trailing CRs from patch.)
patching file src/java/org/apache/hama/bsp/BooleanMessage.java
(Stripping trailing CRs from patch.)
patching file src/examples/org/apache/hama/examples/ExampleDriver.java
(Stripping trailing CRs from patch.)
patching file src/java/org/apache/hama/bsp/IntegerMessage.java
(Stripping trailing CRs from patch.)
patching file src/examples/org/apache/hama/examples/ShortestPaths.java

So carriage returns are causing timeouts in the second run of the test target.
Just with svn diff.

FYI, http://wiki.apache.org/hama/HowToContribute - It'll be helpful for you.

Thanks. But I think this is just strange.

v10 is just a test for me

scheduling build

-1 overall. Here are the results of testing the latest attachment
http://issues.apache.org/jira/secure/attachment/12478509/HAMA-359-v10.patch
against trunk revision 1100148.

@author +1. The patch does not contain any @author tags.

tests included -1. The patch doesn't appear to include any new or modified tests.
Please justify why no tests are needed for this patch.

core tests +1. The patch passed core unit tests.

Changes : http://builds.apache.org/hudson/job/Hama-Patch/334/changes/
Console output: http://builds.apache.org/hudson/job/Hama-Patch/334/console

This message is automatically generated.

Okay, I blame windows.

Thank you for your help.

Added wiki pages for that, would be great if you can commit this.
You've added your own patch then

And not related to this here: what should we do with PageRank?

This patch works only on single node.

21:33,605 INFO org.apache.hadoop.ipc.Server: IPC Server handler 7 on 49258: starting
2011-05-09 13:21:33,605 INFO org.apache.hadoop.ipc.Server: IPC Server handler 8 on 49258: starting
2011-05-09 13:21:33,605 INFO org.apache.hadoop.ipc.Server: IPC Server handler 9 on 49258: starting
2011-05-09 13:21:33,605 INFO org.apache.hama.bsp.GroomServer: GroomServer up at: localhost/127.0.0.1:49258
2011-05-09 13:21:33,605 INFO org.apache.hama.bsp.GroomServer: Starting groom: groomd_cnode2.cloud_61000
2011-05-09 13:21:38,416 INFO org.apache.hama.bsp.TaskRunner: attempt_201105091307_0001_000001_0 11/05/09 13:21:38 WARN util.NativeCodeLoader: Unable t                 o load native-hadoop library for your platform... using builtin-java classes where applicable
2011-05-09 13:21:38,416 INFO org.apache.hama.bsp.TaskRunner: attempt_201105091307_0001_000001_0 11/05/09 13:21:38 INFO compress.CodecPool: Got brand-n                 ew decompressor
2011-05-09 13:21:38,671 INFO org.apache.hama.bsp.TaskRunner: attempt_201105091307_0001_000001_0 11/05/09 13:21:38 WARN bsp.GroomServer: Error running                  child
2011-05-09 13:21:38,671 INFO org.apache.hama.bsp.TaskRunner: attempt_201105091307_0001_000001_0 java.lang.NullPointerException
2011-05-09 13:21:38,671 INFO org.apache.hama.bsp.TaskRunner: attempt_201105091307_0001_000001_0         at org.apache.hama.examples.ShortestPaths.bsp(                 ShortestPaths.java:92)
2011-05-09 13:21:38,672 INFO org.apache.hama.bsp.TaskRunner: attempt_201105091307_0001_000001_0         at org.apache.hama.bsp.BSPTask.run(BSPTask.jav                 a:54)
2011-05-09 13:21:38,672 INFO org.apache.hama.bsp.TaskRunner: attempt_201105091307_0001_000001_0         at org.apache.hama.bsp.GroomServer$Child.main(                 GroomServer.java:849)

How many nodes did you tried?
For me this is working with current trunk and 2 nodes.

2 ~ 16 nodes are all the same. It seems running on only single node.

And I think, it would be better to move all the codes related with graph data loader to unit test.

Thank you for testing this ;D
And I know exactly why.
Because when distributing the data it will use this modulo function on peer.getAllPeerNames().
Since the ordering is not consistent on every groom, this will cause messages to get on the wrong host and therefore nullpointers.

I gonna fix this now.
Should we put just the data loading into a JUnit test or what is your idea?

Okay fixed version, should work fine now. Maven install package completes without errors, so submitting to hudson.

fixed formatting

Testet it with 2 peers now:

hadoop@raynor:/usr/local$ hama/bin/hama jar /usr/local/hama/hama-0.3.0-examples.jar sssp
Single Source Shortest Path Example:
<Startvertex name> <optional: output path> <optional: path to own adjacency list sequencefile>
Setting default start vertex to "Frankfurt"!
11/05/09 22:38:29 INFO examples.ShortestPaths: Starting data partitioning...
11/05/09 22:38:29 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
11/05/09 22:38:29 INFO compress.CodecPool: Got brand-new compressor
11/05/09 22:38:29 INFO compress.CodecPool: Got brand-new compressor
11/05/09 22:38:29 INFO examples.ShortestPaths: Finished!
11/05/09 22:38:29 INFO bsp.BSPJobClient: Running job: job_201105092237_0001
11/05/09 22:38:32 INFO bsp.BSPJobClient: Current supersteps number: 0
11/05/09 22:38:35 INFO bsp.BSPJobClient: Current supersteps number: 12
11/05/09 22:38:38 INFO bsp.BSPJobClient: The total number of supersteps: 12
Job Finished in 9.465 seconds
-------------------- RESULTS --------------------
11/05/09 22:38:38 INFO compress.CodecPool: Got brand-new decompressor
Erfurt | 385
Augsburg | 415
Nuernberg | 320
Stuttgart | 503
Mannheim | 85
Kassel | 173
Muenchen | 487
Karlsruhe | 165
Frankfurt | 0
Wuerzburg | 217

Wow +1.

it works greatly!! It'll be perfect if you can simplify the codes.

I tried to execute 'adjacencylist.seq2' but error occurred as below:

11/05/11 12:24:01 DEBUG bsp.BSPJobClient: BSPJobClient.submitJobDir: hdfs://hnode15:9000/tmp/hadoop-root/bsp/system/submit_q8gy6m
11/05/11 12:24:01 INFO bsp.BSPJobClient: Running job: job_201105111216_0001
11/05/11 12:24:04 INFO bsp.BSPJobClient: Current supersteps number: 0
11/05/11 12:34:56 INFO bsp.BSPJobClient: Current supersteps number: 1
11/05/11 12:38:56 INFO bsp.BSPJobClient: Current supersteps number: 9
11/05/11 12:38:59 INFO bsp.BSPJobClient: Current supersteps number: 12
11/05/11 12:39:02 INFO bsp.BSPJobClient: Current supersteps number: 15
11/05/11 12:39:38 INFO bsp.BSPJobClient: Current supersteps number: 16
11/05/11 12:39:50 INFO bsp.BSPJobClient: Current supersteps number: 18
11/05/11 12:45:17 INFO bsp.BSPJobClient: Current supersteps number: 19
11/05/11 12:47:17 INFO bsp.BSPJobClient: Current supersteps number: 21
^CYou have new mail in /var/mail/root
root@cnode1:/usr/local/src/hama-trunk# bin/hama job -kill job_201105111216_0001


...
2011-05-11 13:02:48,593 DEBUG org.apache.hama.bsp.BSPPeer: Send bytes (-807337319) to cnode3.cloud:61000
2011-05-11 13:02:48,593 DEBUG org.apache.hama.bsp.BSPPeer: Send bytes (-836579055) to cnode3.cloud:61000
2011-05-11 13:02:48,594 DEBUG org.apache.hama.bsp.BSPPeer: Send bytes (-972443212) to cnode13.cloud:61000
2011-05-11 13:02:48,594 DEBUG org.apache.hama.bsp.BSPPeer: Send bytes (-437982308) to cnode2.cloud:61000
2011-05-11 13:02:48,594 DEBUG org.apache.hama.bsp.BSPPeer: Send bytes (-837274598) to cnode4.cloud:61000
2011-05-11 13:02:48,595 DEBUG org.apache.hama.bsp.BSPPeer: [cnode1.cloud:61000] enter the enterbarrier
2011-05-11 13:06:03,316 INFO org.apache.hadoop.ipc.Server: IPC Server handler 5 on 54065, call sync() from 127.0.0.1:45011: error: java.io.IOException: Call to cnode2.cloud/10.5.16.130:61000 failed on local exception: java.io.EOFException
java.io.IOException: Call to cnode2.cloud/10.5.16.130:61000 failed on local exception: java.io.EOFException
        at org.apache.hadoop.ipc.Client.wrapException(Client.java:775)
        at org.apache.hadoop.ipc.Client.call(Client.java:743)
        at org.apache.hadoop.ipc.RPC$Invoker.invoke(RPC.java:220)
        at $Proxy7.put(Unknown Source)
        at org.apache.hama.bsp.BSPPeer.sync(BSPPeer.java:192)
        at org.apache.hama.bsp.GroomServer.sync(GroomServer.java:924)
        at sun.reflect.GeneratedMethodAccessor7.invoke(Unknown Source)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
        at java.lang.reflect.Method.invoke(Method.java:597)
        at org.apache.hadoop.ipc.RPC$Server.call(RPC.java:508)
        at org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:959)
        at org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:955)
        at java.security.AccessController.doPrivileged(Native Method)
        at javax.security.auth.Subject.doAs(Subject.java:396)
        at org.apache.hadoop.ipc.Server$Handler.run(Server.java:953)
Caused by: java.io.EOFException
        at java.io.DataInputStream.readInt(DataInputStream.java:375)
        at org.apache.hadoop.ipc.Client$Connection.receiveResponse(Client.java:501)
        at org.apache.hadoop.ipc.Client$Connection.run(Client.java:446)
2011-05-11 13:06:03,328 INFO org.apache.hama.bsp.TaskRunner: attempt_201105111216_0001_000015_0 org.apache.hadoop.ipc.RemoteException: java.io.IOException: Call to cnode2.cloud/10.5.16.130:61000 failed on local exception: java.io.EOFException
2011-05-11 13:06:03,328 INFO org.apache.hama.bsp.TaskRunner: attempt_201105111216_0001_000015_0         at org.apache.hadoop.ipc.Client.wrapException(Client.java:775)
2011-05-11 13:06:03,328 INFO org.apache.hama.bsp.TaskRunner: attempt_201105111216_0001_000015_0         at org.apache.hadoop.ipc.Client.call(Client.java:743)
2011-05-11 13:06:03,329 INFO org.apache.hama.bsp.TaskRunner: attempt_201105111216_0001_000015_0         at org.apache.hadoop.ipc.RPC$Invoker.invoke(RPC.java:220)
2011-05-11 13:06:03,329 INFO org.apache.hama.bsp.TaskRunner: attempt_201105111216_0001_000015_0         at $Proxy7.put(Unknown Source)
2011-05-11 13:06:03,329 INFO org.apache.hama.bsp.TaskRunner: attempt_201105111216_0001_000015_0         at org.apache.hama.bsp.BSPPeer.sync(BSPPeer.java:192)
2011-05-11 13:06:03,329 INFO org.apache.hama.bsp.TaskRunner: attempt_201105111216_0001_000015_0         at org.apache.hama.bsp.GroomServer.sync(GroomServer.java:924)
2011-05-11 13:06:03,329 INFO org.apache.hama.bsp.TaskRunner: attempt_201105111216_0001_000015_0         at sun.reflect.GeneratedMethodAccessor7.invoke(Unknown Source)
2011-05-11 13:06:03,329 INFO org.apache.hama.bsp.TaskRunner: attempt_201105111216_0001_000015_0         at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
2011-05-11 13:06:03,329 INFO org.apache.hama.bsp.TaskRunner: attempt_201105111216_0001_000015_0         at java.lang.reflect.Method.invoke(Method.java:597)
2011-05-11 13:06:03,329 INFO org.apache.hama.bsp.TaskRunner: attempt_201105111216_0001_000015_0         at org.apache.hadoop.ipc.RPC$Server.call(RPC.java:508)
2011-05-11 13:06:03,329 INFO org.apache.hama.bsp.TaskRunner: attempt_201105111216_0001_000015_0         at org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:959)
2011-05-11 13:06:03,329 INFO org.apache.hama.bsp.TaskRunner: attempt_201105111216_0001_000015_0         at org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:955)
2011-05-11 13:06:03,329 INFO org.apache.hama.bsp.TaskRunner: attempt_201105111216_0001_000015_0         at java.security.AccessController.doPrivileged(Native Method)
2011-05-11 13:06:03,329 INFO org.apache.hama.bsp.TaskRunner: attempt_201105111216_0001_000015_0         at javax.security.auth.Subject.doAs(Subject.java:396)
2011-05-11 13:06:03,329 INFO org.apache.hama.bsp.TaskRunner: attempt_201105111216_0001_000015_0         at org.apache.hadoop.ipc.Server$Handler.run(Server.java:953)
2011-05-11 13:06:03,329 INFO org.apache.hama.bsp.TaskRunner: attempt_201105111216_0001_000015_0 Caused by: java.io.EOFException
2011-05-11 13:06:03,329 INFO org.apache.hama.bsp.TaskRunner: attempt_201105111216_0001_000015_0         at java.io.DataInputStream.readInt(DataInputStream.java:375)
2011-05-11 13:06:03,329 INFO org.apache.hama.bsp.TaskRunner: attempt_201105111216_0001_000015_0         at org.apache.hadoop.ipc.Client$Connection.receiveResponse(Client.java:501)
2011-05-11 13:06:03,329 INFO org.apache.hama.bsp.TaskRunner: attempt_201105111216_0001_000015_0         at org.apache.hadoop.ipc.Client$Connection.run(Client.java:446)
2011-05-11 13:06:03,329 INFO org.apache.hama.bsp.TaskRunner: attempt_201105111216_0001_000015_0
2011-05-11 13:06:03,329 INFO org.apache.hama.bsp.TaskRunner: attempt_201105111216_0001_000015_0         at org.apache.hadoop.ipc.Client.call(Client.java:740)
2011-05-11 13:06:03,329 INFO org.apache.hama.bsp.TaskRunner: attempt_201105111216_0001_000015_0         at org.apache.hadoop.ipc.RPC$Invoker.invoke(RPC.java:220)
2011-05-11 13:06:03,329 INFO org.apache.hama.bsp.TaskRunner: attempt_201105111216_0001_000015_0         at $Proxy0.sync(Unknown Source)
2011-05-11 13:06:03,330 INFO org.apache.hama.bsp.TaskRunner: attempt_201105111216_0001_000015_0         at org.apache.hama.examples.ShortestPaths.bsp(ShortestPaths.java:97)
2011-05-11 13:06:03,330 INFO org.apache.hama.bsp.TaskRunner: attempt_201105111216_0001_000015_0         at org.apache.hama.bsp.BSPTask.run(BSPTask.java:54)
2011-05-11 13:06:03,330 INFO org.apache.hama.bsp.TaskRunner: attempt_201105111216_0001_000015_0         at org.apache.hama.bsp.GroomServer$Child.main(GroomServer.java:849)

Thank you
I have an exam tomorrow, I'll try to look at it afterwards.

Did you looked into the logs? I've added some status outputs, it is possible that the hosts were just writing the results to disk, this takes a while. From the main method it looks like the whole job is frozen.

After you've killed it, you have seen that the code was just syncing.
How much memory did you used per task? I have to allocate about 10gb. Java HashMap's have a lot of overhead.
Maybe it hang in GC too much, that's what I've seen during profiling too.

P.S. the error always occurs at the 21th superstep.

That is strange, I have to profile this. How many grooms did you used?

11/05/11 08:08:34 INFO bsp.BSPJobClient: Current supersteps number: 17
11/05/11 08:08:37 INFO bsp.BSPJobClient: Current supersteps number: 18
11/05/11 08:08:57 INFO bsp.BSPJobClient: Current supersteps number: 20
11/05/11 08:09:31 INFO bsp.BSPJobClient: Current supersteps number: 21
11/05/11 08:09:54 INFO bsp.BSPJobClient: Current supersteps number: 23
11/05/11 08:10:12 INFO bsp.BSPJobClient: Current supersteps number: 24
11/05/11 08:10:15 INFO bsp.BSPJobClient: Current supersteps number: 26
11/05/11 08:10:18 INFO bsp.BSPJobClient: Current supersteps number: 68
11/05/11 08:10:24 INFO bsp.BSPJobClient: The total number of supersteps: 68
Job Finished in 236.995 seconds

Maybe you have to give the tasks a bit more of memory

Besides the hanging the adjacency file has too high weights so when calculating the integer have an overflow.
I'll generate a new one.

The new one workes fine for me. Used a upperlimit of 50k for the weight, so there shouldn't be any overflows.
The costs seems reasonable then. Uploading should take an hour or two, I'll tell you when it is finished.

During that time I investigate why the stuff is hanging.

16 nodes (256 cores), 10G network, -Xmx10000m

wow nice cluster. So this won't be the problem ;D
I have to setup a few more VM's and then I profile this.

New file is downloadable. I have to download it to my VM's now, and then I can profile it.

FYI, below initialization part takes ±10 minutes on 16 physical machines. Is there a reason why use sendMessageToNeighbors() method?

+    // init the vertices
+    for (ShortestPathVertex v : adjacencyList.keySet()) {
+      if (v.equals(start)) {
+        v.setCost(0);
+      } else {
+        // INF
+        v.setCost(Integer.MAX_VALUE);
+      }
+      // initial message bypass
+      sendMessageToNeighbors(peer, v);
+    }

We can initilize the vertex while we read the map, if the groom contains the start vertex it can send the message to the neighbors.
I'll fix this.

BTW partitioning takes much time as well. Maybe we can start a MapReduce job that distribute the keys to their buckets.. This would parallize the whole step in some way.

*add init phase to the memory mapping
*remove the init overhead: just trigger the first superstep with the startvertex

Thanks for the suggestion!

Did you already tried the new file + init fix?

So I roughly fixed my VMs but it is still a 2 node cluster. But profiling should be able to determine the problems

Mem usage on 64 for the sample set is about 1,9gb for each host (of two).

Everything runs fine. Profiling exposes a spending of 40% time in Connection.run() -> receiveResponse().
And 40% arround mapping the stuff from SequenceFiles into RAM.
Partitioning takes time as well... So I think running a mapreduce job in front of the task is a good idea. That is a bit tricky because of hadoop's hashpartitioner, and BSP picks the right outputfile... Maybe I can find a way to exploit this ;D

Never the less, GC shouldn't be the problem why your tasks don't make any progress.

Now on 6 Nodes:

Now I faced the haging with the standard example. But that is really random, It's in 1 of 5 cases...
And the same with bench 512 50 1000. Even during:

hama/bin/hama jar /usr/local/hama/hama-0.3.0-examples.jar test
11/05/12 23:48:12 INFO bsp.BSPJobClient: Running job: job_201105122340_0003
11/05/12 23:48:15 INFO bsp.BSPJobClient: Current supersteps number: 0
11/05/12 23:48:18 INFO bsp.BSPJobClient: Current supersteps number: 2
... [Hangs forever]

Could this be a problem of fault tolerance? Like one groom isn't responsive anymore and the others are just waiting for this groom to reach the barrier?

Another observation, in the long sequencefile SSSP some crucial slowdowns occus between Superstep 15 and 18.
Using TCPDUMP shows that there is actually a communication, but with a lot of small packets (length 12 - 20)and it takes a long time.

11/05/13 00:01:32 INFO bsp.BSPJobClient: Running job: job_201105122350_0004
11/05/13 00:01:35 INFO bsp.BSPJobClient: Current supersteps number: 0
11/05/13 00:02:44 INFO bsp.BSPJobClient: Current supersteps number: 9
11/05/13 00:02:53 INFO bsp.BSPJobClient: Current supersteps number: 10
11/05/13 00:02:56 INFO bsp.BSPJobClient: Current supersteps number: 12
11/05/13 00:03:50 INFO bsp.BSPJobClient: Current supersteps number: 13
11/05/13 00:04:17 INFO bsp.BSPJobClient: Current supersteps number: 15
11/05/13 00:14:36 INFO bsp.BSPJobClient: Current supersteps number: 16
11/05/13 00:19:30 INFO bsp.BSPJobClient: Current supersteps number: 18

Picked the communication between two of six nodes:

00:20:42.160759 IP raynor.21810 > zeratul.37758: Flags [P.], seq 1:21, ack 12, win 46, options [nop,nop,TS val 501715 ecr 304403], length 20
00:20:42.160928 IP zeratul.37758 > raynor.21810: Flags [.], ack 21, win 501, options [nop,nop,TS val 304403 ecr 501715], length 0
00:20:44.169980 IP zeratul.37758 > raynor.21810: Flags [P.], seq 12:24, ack 21, win 501, options [nop,nop,TS val 304604 ecr 501715], length 12
00:20:44.170347 IP raynor.21810 > zeratul.37758: Flags [P.], seq 21:41, ack 24, win 46, options [nop,nop,TS val 501916 ecr 304604], length 20
00:20:44.170748 IP zeratul.37758 > raynor.21810: Flags [.], ack 41, win 501, options [nop,nop,TS val 304604 ecr 501916], length 0
00:20:46.170129 IP zeratul.37758 > raynor.21810: Flags [P.], seq 24:36, ack 41, win 501, options [nop,nop,TS val 304804 ecr 501916], length 12
00:20:46.170867 IP raynor.21810 > zeratul.37758: Flags [P.], seq 41:61, ack 36, win 46, options [nop,nop,TS val 502116 ecr 304804], length 20
00:20:46.171227 IP zeratul.37758 > raynor.21810: Flags [.], ack 61, win 501, options [nop,nop,TS val 304804 ecr 502116], length 0
00:20:48.170054 IP zeratul.37758 > raynor.21810: Flags [P.], seq 36:48, ack 61, win 501, options [nop,nop,TS val 305004 ecr 502116], length 12
00:20:48.170536 IP raynor.21810 > zeratul.37758: Flags [P.], seq 61:81, ack 48, win 46, options [nop,nop,TS val 502316 ecr 305004], length 20
00:20:48.170959 IP zeratul.37758 > raynor.21810: Flags [.], ack 81, win 501, options [nop,nop,TS val 305004 ecr 502316], length 0

Solution:
They messages are in a queue, so I implement a list writable that will batch these vertices together. I hope this will result in a far better running time...
We should take this experience into consideration for the other GSOC task.

>> Could this be a problem of fault tolerance? Like one groom isn't responsive anymore and the others are just waiting for this groom to reach the barrier?

What's the error log say? it seems similar with HAMA-355 problem.

and, I tried to run with new patch, the EOF (readInt() call) error still occurs at 21th superstep.

That is really strange, I investigate further ;D

It is getting even funnier

11/05/13 11:18:52 INFO bsp.BSPJobClient: Running job: job_201105131031_0003
11/05/13 11:18:55 INFO bsp.BSPJobClient: Current supersteps number: 0
11/05/13 11:20:01 INFO bsp.BSPJobClient: Current supersteps number: 1
11/05/13 11:20:04 INFO bsp.BSPJobClient: Current supersteps number: 0
11/05/13 11:20:07 INFO bsp.BSPJobClient: Current supersteps number: 1
11/05/13 11:20:10 INFO bsp.BSPJobClient: Current supersteps number: 0
11/05/13 11:20:13 INFO bsp.BSPJobClient: Current supersteps number: 1
11/05/13 11:20:16 INFO bsp.BSPJobClient: Current supersteps number: 0
11/05/13 11:20:19 INFO bsp.BSPJobClient: Current supersteps number: 1
11/05/13 11:20:22 INFO bsp.BSPJobClient: Current supersteps number: 0
11/05/13 11:20:25 INFO bsp.BSPJobClient: Current supersteps number: 1
11/05/13 11:20:28 INFO bsp.BSPJobClient: Current supersteps number: 0
11/05/13 11:20:31 INFO bsp.BSPJobClient: Current supersteps number: 1
11/05/13 11:20:34 INFO bsp.BSPJobClient: Current supersteps number: 0
11/05/13 11:20:37 INFO bsp.BSPJobClient: Current supersteps number: 1

I don't know what is going wrong...

• added a partitioning skip flag
• debug outputs

2011-05-13 12:55:14,227 INFO org.apache.hadoop.ipc.Server: IPC Server handler 7 on 56482, call sync() from 127.0.0.
1:58149: error: java.io.IOException: org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode
 for /bsp/horner:61000
java.io.IOException: org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /bsp/horner
:61000
        at org.apache.zookeeper.KeeperException.create(KeeperException.java:102)
        at org.apache.zookeeper.KeeperException.create(KeeperException.java:42)
        at org.apache.zookeeper.ZooKeeper.delete(ZooKeeper.java:728)
        at org.apache.hama.bsp.BSPPeer.leaveBarrier(BSPPeer.java:262)
        at org.apache.hama.bsp.BSPPeer.sync(BSPPeer.java:203)
        at org.apache.hama.bsp.GroomServer.sync(GroomServer.java:924)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
        at java.lang.reflect.Method.invoke(Method.java:597)
        at org.apache.hadoop.ipc.RPC$Server.call(RPC.java:508)
        at org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:961)
        at org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:957)
        at java.security.AccessController.doPrivileged(Native Method)
        at javax.security.auth.Subject.doAs(Subject.java:396)
        at org.apache.hadoop.ipc.Server$Handler.run(Server.java:955)

You're right with the barrier.

2011-05-16 18:01:46,251 INFO org.apache.zookeeper.server.PrepRequestProcessor: Got user-level KeeperException when processing sessionid:0x12ff96fd5460004 type:create cxid:0x45 zxid:0xfffffffffffffffe txntype:unknown reqpath:n/a Error Path:/bsp/findlay:61000 Error:KeeperErrorCode = NodeExists for /bsp/findlay:61000
2011-05-16 18:03:31,892 INFO org.apache.zookeeper.server.PrepRequestProcessor: Got user-level KeeperException when processing sessionid:0x12ff96fd5460001 type:create cxid:0x5b zxid:0xfffffffffffffffe txntype:unknown reqpath:n/a Error Path:/bsp/raynor:61000 Error:KeeperErrorCode = NodeExists for /bsp/raynor:61000
2011-05-16 18:03:32,256 INFO org.apache.zookeeper.server.PrepRequestProcessor: Got user-level KeeperException when processing sessionid:0x12ff96fd5460004 type:create cxid:0x4c zxid:0xfffffffffffffffe txntype:unknown reqpath:n/a Error Path:/bsp/findlay:61000-data Error:KeeperErrorCode = NodeExists for /bsp/findlay:61000-data
2011-05-16 18:03:32,407 INFO org.apache.zookeeper.server.PrepRequestProcessor: Got user-level KeeperException when processing sessionid:0x12ff96fd5460001 type:create cxid:0x63 zxid:0xfffffffffffffffe txntype:unknown reqpath:n/a Error Path:/bsp/raynor:61000-data Error:KeeperErrorCode = NodeExists for /bsp/raynor:61000-data
2011-05-16 18:03:32,448 INFO org.apache.zookeeper.server.PrepRequestProcessor: Got user-level KeeperException when processing sessionid:0x12ff96fd5460004 type:create cxid:0x51 zxid:0xfffffffffffffffe txntype:unknown reqpath:n/a Error Path:/bsp/findlay:61000 Error:KeeperErrorCode = NodeExists for /bsp/findlay:61000
2011-05-16 18:03:32,454 INFO org.apache.zookeeper.server.PrepRequestProcessor: Got user-level KeeperException when processing sessionid:0x12ff96fd5460004 type:create cxid:0x53 zxid:0xfffffffffffffffe txntype:unknown reqpath:n/a Error Path:/bsp/findlay:61000-data Error:KeeperErrorCode = NodeExists for /bsp/findlay:61000-data
2011-05-16 18:03:32,523 INFO org.apache.zookeeper.server.PrepRequestProcessor: Got user-level KeeperException when processing sessionid:0x12ff96fd5460005 type:delete cxid:0x66 zxid:0xfffffffffffffffe txntype:unknown reqpath:n/a Error Path:/bsp/horner:61000 Error:KeeperErrorCode = NoNode for /bsp/horner:61000

These are Zookeeper logs, that seems to be the problem. Job hangs until it gets killed.

This here looks like a BIG flaw, this happens when killing the job.

2011-05-16 17:32:07,230 INFO org.apache.hama.bsp.GroomServer: Lost connection to BSP Master [raynor/192.168.1.1:9002].  Retrying...
java.util.ConcurrentModificationException
        at java.util.LinkedHashMap$LinkedHashIterator.nextEntry(LinkedHashMap.java:373)
        at java.util.LinkedHashMap$EntryIterator.next(LinkedHashMap.java:392)
        at java.util.LinkedHashMap$EntryIterator.next(LinkedHashMap.java:391)
        at org.apache.hama.bsp.GroomServer.offerService(GroomServer.java:374)
        at org.apache.hama.bsp.GroomServer.run(GroomServer.java:609)
        at java.lang.Thread.run(Thread.java:662)

That is the client side exception

2011-05-16 18:07:23,583 INFO org.apache.hadoop.ipc.Server: IPC Server handler 6 on 46733, call sync() from 127.0.0.1:50305: error: java.io.IOException: org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /bsp/findlay:61000
java.io.IOException: org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /bsp/findlay:61000
        at org.apache.zookeeper.KeeperException.create(KeeperException.java:102)
        at org.apache.zookeeper.KeeperException.create(KeeperException.java:42)
        at org.apache.zookeeper.ZooKeeper.delete(ZooKeeper.java:728)
        at org.apache.hama.bsp.BSPPeer.leaveBarrier(BSPPeer.java:262)
        at org.apache.hama.bsp.BSPPeer.sync(BSPPeer.java:203)
        at org.apache.hama.bsp.GroomServer.sync(GroomServer.java:924)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
        at java.lang.reflect.Method.invoke(Method.java:597)
        at org.apache.hadoop.ipc.RPC$Server.call(RPC.java:508)
        at org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:961)
        at org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:957)
        at java.security.AccessController.doPrivileged(Native Method)
        at javax.security.auth.Subject.doAs(Subject.java:396)
        at org.apache.hadoop.ipc.Server$Handler.run(Server.java:955)

Do you know a valid fix for this?

I think we should create a task to refactor the exception.printStackTrace() to LOG.error(exception,"msg").

>> I think we should create a task to refactor the exception.printStackTrace() to LOG.error(exception,"msg").

+1

BTW, file loading and partitioning job are too heavy as you said. for simple debugging, could you please upload the partitioned files e.g., part-1, part-2, ... part-16?

Yep, I'm currently evaluating a MapReduce job that makes the partitioning.

The problem with the partitioning itself is that the parts are named like the grooms.
But I can overload the partition method once again and deal with part-X and just rename it.

I open a task for the error logs.

My problem has been solved by increasing the 'HAMA_HEAPSIZE' (to 10g) and the 'zookeeper.session.timeout' (to 1200000). It was some problem of GroomServers.

But, ...

11/05/18 14:05:14 INFO bsp.BSPJobClient: Running job: job_201105181357_0002
11/05/18 14:05:17 INFO bsp.BSPJobClient: Current supersteps number: 0
11/05/18 14:05:26 INFO bsp.BSPJobClient: Current supersteps number: 5
11/05/18 14:05:29 INFO bsp.BSPJobClient: Current supersteps number: 10
11/05/18 14:05:32 INFO bsp.BSPJobClient: Current supersteps number: 12
11/05/18 14:05:35 INFO bsp.BSPJobClient: Current supersteps number: 13
11/05/18 14:05:38 INFO bsp.BSPJobClient: Current supersteps number: 15
11/05/18 14:06:14 INFO bsp.BSPJobClient: Current supersteps number: 16
11/05/18 14:06:26 INFO bsp.BSPJobClient: Current supersteps number: 18
11/05/18 14:11:56 INFO bsp.BSPJobClient: Current supersteps number: 19
11/05/18 14:13:56 INFO bsp.BSPJobClient: Current supersteps number: 21
11/05/18 14:33:21 INFO bsp.BSPJobClient: Current supersteps number: 22
11/05/18 14:40:09 INFO bsp.BSPJobClient: Current supersteps number: 24
11/05/18 14:49:33 INFO bsp.BSPJobClient: Current supersteps number: 25
11/05/18 14:53:03 INFO bsp.BSPJobClient: Current supersteps number: 27
11/05/18 14:56:24 INFO bsp.BSPJobClient: Current supersteps number: 28
11/05/18 15:09:28 INFO bsp.BSPJobClient: Current supersteps number: 27
11/05/18 15:09:31 INFO bsp.BSPJobClient: Current supersteps number: 28
11/05/18 15:09:34 INFO bsp.BSPJobClient: Current supersteps number: 27
11/05/18 15:09:37 INFO bsp.BSPJobClient: Current supersteps number: 28
11/05/18 15:09:40 INFO bsp.BSPJobClient: Current supersteps number: 27
11/05/18 15:09:43 INFO bsp.BSPJobClient: Current supersteps number: 28
11/05/18 15:09:46 INFO bsp.BSPJobClient: Current supersteps number: 27
11/05/18 15:09:49 INFO bsp.BSPJobClient: Current supersteps number: 28
11/05/18 15:09:52 INFO bsp.BSPJobClient: Current supersteps number: 27
11/05/18 15:09:55 INFO bsp.BSPJobClient: Current supersteps number: 28
11/05/18 15:09:58 INFO bsp.BSPJobClient: Current supersteps number: 27
11/05/18 15:10:01 INFO bsp.BSPJobClient: Current supersteps number: 28
11/05/18 15:10:04 INFO bsp.BSPJobClient: Current supersteps number: 27
11/05/18 15:10:07 INFO bsp.BSPJobClient: Current supersteps number: 28
11/05/18 15:10:10 INFO bsp.BSPJobClient: Current supersteps number: 27
11/05/18 15:10:13 INFO bsp.BSPJobClient: Current supersteps number: 28
11/05/18 15:10:16 INFO bsp.BSPJobClient: Current supersteps number: 27
11/05/18 15:10:19 INFO bsp.BSPJobClient: Current supersteps number: 28
11/05/18 15:10:22 INFO bsp.BSPJobClient: Current supersteps number: 27
...

Typo: some problem -> OOME problem

That was obvious, the tasks get executed in the groom.
I don't really know what we can do against this superstep "hopping". Is this still related to the barrier sync? Increasing the timeout won't fix the problem with it.

I tested after changing 'Thread.sleep(100);' to 'Thread.sleep(10000);' at BSPPeer.sync() method and finally, my job successfully done.

2011-05-18 15:15:27,147 INFO org.apache.hadoop.ipc.Server: IPC Server listener on 40000: starting
2011-05-18 15:15:27,149 INFO org.apache.hadoop.ipc.Server: IPC Server handler 0 on 40000: starting
2011-05-18 15:15:27,151 INFO org.apache.hama.bsp.BSPMaster: Starting RUNNING
2011-05-18 15:22:48,062 DEBUG org.apache.hama.bsp.JobInProgress: numBSPTasks: 16
2011-05-18 15:22:48,065 DEBUG org.apache.hama.bsp.JobInProgress: Job is initialized.
2011-05-18 16:29:27,582 INFO org.apache.hama.bsp.JobInProgress: Taskid 'attempt_201105181515_0001_000005_0' has finished successfully.
2011-05-18 16:29:27,583 INFO org.apache.hama.bsp.TaskInProgress: Task 'task_201105181515_0001_000005' has completed.
2011-05-18 16:29:27,806 INFO org.apache.hama.bsp.JobInProgress: Taskid 'attempt_201105181515_0001_000004_0' has finished successfully.
2011-05-18 16:29:27,806 INFO org.apache.hama.bsp.TaskInProgress: Task 'task_201105181515_0001_000004' has completed.
2011-05-18 16:29:28,336 INFO org.apache.hama.bsp.JobInProgress: Taskid 'attempt_201105181515_0001_000015_0' has finished successfully.
2011-05-18 16:29:28,336 INFO org.apache.hama.bsp.TaskInProgress: Task 'task_201105181515_0001_000015' has completed.
2011-05-18 16:29:28,517 INFO org.apache.hama.bsp.JobInProgress: Taskid 'attempt_201105181515_0001_000000_0' has finished successfully.
2011-05-18 16:29:28,517 INFO org.apache.hama.bsp.TaskInProgress: Task 'task_201105181515_0001_000000' has completed.
2011-05-18 16:29:28,524 INFO org.apache.hama.bsp.JobInProgress: Taskid 'attempt_201105181515_0001_000013_0' has finished successfully.
2011-05-18 16:29:28,524 INFO org.apache.hama.bsp.TaskInProgress: Task 'task_201105181515_0001_000013' has completed.
2011-05-18 16:29:28,589 INFO org.apache.hama.bsp.JobInProgress: Taskid 'attempt_201105181515_0001_000012_0' has finished successfully.
2011-05-18 16:29:28,589 INFO org.apache.hama.bsp.TaskInProgress: Task 'task_201105181515_0001_000012' has completed.
2011-05-18 16:29:28,602 INFO org.apache.hama.bsp.JobInProgress: Taskid 'attempt_201105181515_0001_000001_0' has finished successfully.
2011-05-18 16:29:28,602 INFO org.apache.hama.bsp.TaskInProgress: Task 'task_201105181515_0001_000001' has completed.
2011-05-18 16:29:28,775 INFO org.apache.hama.bsp.JobInProgress: Taskid 'attempt_201105181515_0001_000014_0' has finished successfully.
2011-05-18 16:29:28,775 INFO org.apache.hama.bsp.TaskInProgress: Task 'task_201105181515_0001_000014' has completed.
2011-05-18 16:29:28,909 INFO org.apache.hama.bsp.JobInProgress: Taskid 'attempt_201105181515_0001_000010_0' has finished successfully.
2011-05-18 16:29:28,909 INFO org.apache.hama.bsp.TaskInProgress: Task 'task_201105181515_0001_000010' has completed.
2011-05-18 16:29:28,914 INFO org.apache.hama.bsp.JobInProgress: Taskid 'attempt_201105181515_0001_000007_0' has finished successfully.
2011-05-18 16:29:28,914 INFO org.apache.hama.bsp.TaskInProgress: Task 'task_201105181515_0001_000007' has completed.
2011-05-18 16:29:28,960 INFO org.apache.hama.bsp.JobInProgress: Taskid 'attempt_201105181515_0001_000006_0' has finished successfully.
2011-05-18 16:29:28,960 INFO org.apache.hama.bsp.TaskInProgress: Task 'task_201105181515_0001_000006' has completed.
2011-05-18 16:29:29,148 INFO org.apache.hama.bsp.JobInProgress: Taskid 'attempt_201105181515_0001_000011_0' has finished successfully.
2011-05-18 16:29:29,148 INFO org.apache.hama.bsp.TaskInProgress: Task 'task_201105181515_0001_000011' has completed.
2011-05-18 16:29:29,199 INFO org.apache.hama.bsp.JobInProgress: Taskid 'attempt_201105181515_0001_000008_0' has finished successfully.
2011-05-18 16:29:29,199 INFO org.apache.hama.bsp.TaskInProgress: Task 'task_201105181515_0001_000008' has completed.
2011-05-18 16:29:29,244 INFO org.apache.hama.bsp.JobInProgress: Taskid 'attempt_201105181515_0001_000009_0' has finished successfully.
2011-05-18 16:29:29,244 INFO org.apache.hama.bsp.TaskInProgress: Task 'task_201105181515_0001_000009' has completed.
2011-05-18 16:29:29,274 INFO org.apache.hama.bsp.JobInProgress: Taskid 'attempt_201105181515_0001_000002_0' has finished successfully.
2011-05-18 16:29:29,274 INFO org.apache.hama.bsp.TaskInProgress: Task 'task_201105181515_0001_000002' has completed.
2011-05-18 16:29:29,392 INFO org.apache.hama.bsp.JobInProgress: Taskid 'attempt_201105181515_0001_000003_0' has finished successfully.
2011-05-18 16:29:29,392 INFO org.apache.hama.bsp.TaskInProgress: Task 'task_201105181515_0001_000003' has completed.
2011-05-18 16:29:29,395 DEBUG org.apache.hama.bsp.JobInProgress: Job successfully done.
2011-05-18 16:32:48,365 DEBUG org.apache.hama.bsp.BSPMaster: returns all jobs: 1

>> Is this still related to the barrier sync?

Yes. the problem related with zk node creation/deletion logic in enterBarrier() and leaveBarrier() methods. Sometimes they occurs at the same time.

>> Increasing the timeout won't fix the problem with it.

As i mentioned on chat, JVM garbage collection pause causes zk session time-out errors.

Obviously the task gets executed in a seperate VM. Sorry for that wrong stuff ;D
I'm tracking the GC problem now.

The task itself is fine, it just consumes 700mb.

To profile the others we need to add this line to hama/hama:

HAMA_OPTS="$HAMA_OPTS -agentpath:/usr/local/yk/libyjpagent.so"

SuperStep 15:
Each groom has 50mb allocated and arround 300k integermessages in ram.

At this point there is no problem ;D

Okay so I've seen the problem during superstep 23, a whole lot of messages are coming in and there wasn't enough heap to fit in causing the whole process to get a big GC pause.
Sadly I coudn't get any snapshots because the profiler crashed.

I guess, it could differ by size of splits and number of tasks.

This could be one option.
My approach would be to use an iterator during the message sending phase. (In BSPPeer)
Currently we are just bundling the messages in the MessageBundle class. This gets cleaned up (completely) after the method end, same with outgoing queue which gets cleared afterwards.

We can use iterator.remove to smoothly release references to the objects. This should cause more minor collections instead of one or two larger ones.

Refactored a bit, added an own package and outsourced some utils.

Please review this.

Output is not good.

What's the unit of distance?

...
Horseed | -1911766075
Skol Nakon | -1778351474
Nekatitaranagahawewa | -2046697766
Kimango | -1481513959
Gandoom Ban | -1738053996
Ban Sua Thao | -1569376795
Wanninah al Gharbiyah | -2135610717
El Miadieh | -1685846334
Etehonu | -1680992012
Essong I | -1911876851
Cortijo Corvera | -1867577110
Belomestnaya Dvoynya | -1945614380
Little Packington | -2083576015
Horse Landing | -1930041307
Tagyongni | -1698738295

Did you used the newer sequencefile?

above results are printed by your code.

I'm pretty sure that this is the output of my code.
Did you used the newer version of the SequenceFile? What parameters did you used?

I downloaded again and test is successfully completed on my cluster.

Let's commit this patch first, and later improve the code! What do you think?

Okay
+1

I've committed this. Thanks, Thomas!

Cool thanks. Did you put this into the 3.0 branch?

Yes.

I think, this is not ready to release yet.

The first impression is important!

I agree with you
We have a lot of time to improve.