Worker assignment for apache giraph job

Apache giraph job must be assigned the number of workers(which is basically count of mappers). But with how much workers should we assign for best runtime of our job?
Well few tips I got from my friend semih, from standard university who also developed GPS, an alternative to apache giraph project.

Tips:

Each node should be configured to have mapper count same as number of processor  cores. So if you have 2 core in the processor of your hadoop node than set mapred.map.tasks=2.

Each job must be assigned workers in the multiple of hadoop nodes. Let say you have 8 node cluster than you must assign "-w" value as 7 or 15 or 23 etc.. Note: worker count start from zero(0).

Again thanks to semih, phD scholar from Stanford University for these tips.

Writeup I sent to semih, Stanford University for my Giraph Tester project.

Question: How to write and run your own apache giraph code (Computation / InputFormat / OutputFormat)?

Answer: There are two ways according to my knowledge.

Way 1:

1. Write your custom computation code.

2. Copy that file in apache giraph source code's “giraph-examples” folder.

3. Compile whole giraph source code again with maven.

4. Use giraph jar to run your code, similar to example given on giraph quick start page.

Problem:

Very slow process.

Hadoop psudo-mode setup require.

Input and output files are in HDFS.

Everytime you need to compile giraph-source code with maven.

Way 2:

1. Use Eclipse or other IDE

2. Add jar files from Apache-Hadoop's lib folder & Apache-Giraph's lib folder in Build-path.

3. Write your custom inputFormat/outputFormat/Computation java code

4. Write your giraph Runner java code. (Giraph Job Runner file)

5. Run and test your code on single Click

Advantages:

Faster than Way1.

No Hadoop setup require.

Input and Output files are on local Systems only.

In development phase, we make lot of changes in our code and we need fast result on our sample test input file. So way-2 works better in this case.

Question: How we can debug in those two cases?

Answer: We have already discussed two Ways to write/run giraph code.

For way-1, semih and his team at Stanford University have developed tool called Graft and now it is part of Apache Giraph project.

For way-2, I am not sure if Graft can also work in this case. If not than we can build another debugger to work for way-2.

Question: How to approach for building debugger for Way-2?

Answer: Basic idea is to trace the state of vertice,edges & messages of each superstep, store them in JSON format and plot them using your custom Graph Visulization program.

Question: How much progress I have done for building debugger for way-2?

Answer: I have partially developed Graph Visulization program which is inspired by Graft. I have defined my own JSON format and using it to plot graphs according to coresponding supersteps.

Question: Where Am I stuck?

Answer: I am not able figure out on how to trace program. I must store trace in JSON format and visualize it.

I can either write my trace method which need to be called by user and use will feed current vertex & message status as parameter in it.

I may also change the original java source code of standard files(like BasicComputation) and repackage them in jar. So user must use my modified jar files instead of original giraph-lib jars.

Pregel: Distributed Graph Processing Framework My Notes

Pregel: Distributed Graph Processing Framework

Motivation:-

Efficient processing of large graph is facing following problems.

• Poor locality of memory access
• Very little work per vertex
• Changing degree of parallelism over the course of execution

No scalable general purpose system available for implementing arbitrary graph algorithm over arbitrary graph representation in large-scale distributed environment.

Algorithm implementation to process large graph can be done by one of the following options.

1) Designing custom distributed infrastructure which require considerable implementation effort for each new algorithm or graph representation.

2) Using available distributed computing platform which are not always well suited for graph processing like MapReduce.

3) Use of single computer graph algorithm like BGL, LEDA, NetworkX, JDSL which limit the scalability

4) Using existing parallel graph system like BGL & CGMgraph but they do not address fault tolerances or other distributed system issue.

Proposed Solution:-

Valiant’s Bulk Synchronous Parallel Model

Vertex-Centric approach to solve problem

Pragel computations consist of sequences of iterations called super Supersteps.

During each Superstep S following operations can be performed,

1. It can compute user defined function for each vertex V
2. Each V can read message, sent it while S-1 Superstep.
3. It can send message to V that will be received in S+1 Superstep
4. It can modify state of V & outgoing edge. It can also change graph topology

Note: Messages are typically sent along outgoing edge, but message can be sent to any vertex whose identifier is known.

Computation Model:-

Input: Directed Graph

Each vertex has Vertex-Identifier and modifiable user defined value. Each directed edge has Source Vertices, modifiable user defined value and target vertex identifier.

Pregel computation consists of,

• Input
• Supersteps separated by global synchronization points
• Algorithm termination
• Output

Each vertex compute in parallel with same user defined function.

Algorithm Termination: It terminates when every vertex voting for halt.

Example,

Consider Superstep 0 when all vertexes are active. A vertex deactivates itself by voting for halt. If halted vertex receives any messages then it again activate. To go again in deactivate stage, vertex must vote for halt again. Algorithm terminates when every vertex vote for halt.

Vertex & edge can be added & removed during computations.

Advantage over MapReduce:-

1. Graph algorithm can be written as series of chained MapReduce invocation. This has bad performance & usability. Pregel overcome this problems.
2. Pregel keeps vertices & edges on machine where it performs computation & only use network for message passing. But heavy network bandwidth is used in MapReduce.
3. MapReduce is functional type programming, so expressing graph algorithm as chained MapReduce require passing entire state of the graph from one stage to next stage producing large overhead on communication & associated serializability.
4. There is a need to co-ordinate the steps of chained MapReduce which adds programming complexity. That is avoided by Pregel because of Bulk Synchronization Model.

Sample Problem Solved using Pregel:-

Objective: Find maximum number

Here dotted lines are messages. Dark vertices have voted to halt. In each step supersteps, vertices send maximum vertex value to their neighbor vertices. If vertex itself is bigger than its neighbor vertex then it votes for halt. Algorithm terminates when every vertex halt.

References:-

 [1] G. Malewicz, M. H. Austern, A. J. C. Bik, J. C. Dehnert,I. Horn, N. Leiser, and G. zajkowski. Pregel: A System for Large-Scale Graph Processing. In SIGMOD, 2011.

Giraph Code to find maximum Value of node

Giraph Version:1.1.0
Hadoop Version: 0.20.203

Input Graph:
store it it tiny_graph.txt and upload it on HDFS
[0,5,[[1,1],[3,3]]]
[2,66,[[1,2],[4,4]]]
[1,25,[[0,1],[2,2],[3,1]]]
[3,85,[[0,3],[1,1],[4,4]]]
[4,125,[[3,4],[2,4]]]

Input Graph Pattern:  [VertexId,VertexValue,[EdgeDestinationVertexId,EdgeValue]....]


code:

package Success;
import java.io.IOException;

import org.apache.giraph.graph.BasicComputation;
import org.apache.giraph.graph.Vertex;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.FloatWritable;
import org.apache.hadoop.io.LongWritable;


public class MaxVal extends BasicComputation<LongWritable,DoubleWritable,FloatWritable,DoubleWritable>
 {

    @Override
    public void compute(
            Vertex<LongWritable, DoubleWritable, FloatWritable> v,
            Iterable<DoubleWritable> msg) throws IOException {
        // TODO Auto-generated method stub
        boolean changed=false;
       
        for(DoubleWritable messages :msg)
        {
           
           
            if(v.getValue().get() < messages.get())
            {
                v.setValue(messages);
                changed=true;
            }
        }
       
        if(getSuperstep()==0 | changed)
        {
            sendMessageToAllEdges(v,v.getValue());
        }
       
        v.voteToHalt();
    }

}

How to Compile:

copy this java file to:
$cp your_program.java /usr/local/giraph/giraph-examples/src/main/java/org/apache/giraph/examples/your_program.java

$cd /usr/local/giraph/giraph-examples/
$mvn compile


How to run:

hadoop jar /usr/local/giraph/giraph-examples/target/giraph-examples-1.1.0-SNAPSHOT-for-hadoop-1.2.1-jar-with-dependencies.jar org.apache.giraph.GiraphRunner MaxVal  -vif org.apache.giraph.io.formats.JsonLongDoubleFloatDoubleVertexInputFormat -vof org.apache.giraph.io.formats.JsonLongDoubleFloatDoubleVertexOutputFormat -vip /input/tiny_graph.txt -op /output/MaxVal -w 1

What are arguments?
-vif    :    Graph input format
-vof   :    Graph output format
-vip    :    Graph input file path on HDFS
-op      :    Graph outout path on HDFS

Giraph sendMessageToAllEdges() API

 sendMessageToAllEdges(vertex,data)

This Giraph API send messages to all the vertices directly connected with given vertex.
Tested for undirected graph.