svn commit: r1448524 - in /hama/trunk: ./ examples/src/main/java/org/apache/hama/examples/util/ examples/src/test/java/org/apache/hama/examples/

[tj...@apache.org]

Author: tjungblut
Date: Thu Feb 21 06:51:57 2013
New Revision: 1448524

URL: http://svn.apache.org/r1448524
Log:
[HAMA-736]: Add one pass graph generator

Added:
    hama/trunk/examples/src/main/java/org/apache/hama/examples/util/FastGraphGen.java
    hama/trunk/examples/src/test/java/org/apache/hama/examples/FastGraphGenTest.java
Modified:
    hama/trunk/CHANGES.txt
    hama/trunk/examples/src/main/java/org/apache/hama/examples/util/Generator.java

Modified: hama/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/hama/trunk/CHANGES.txt?rev=1448524&r1=1448523&r2=1448524&view=diff
==============================================================================
--- hama/trunk/CHANGES.txt (original)
+++ hama/trunk/CHANGES.txt Thu Feb 21 06:51:57 2013
@@ -4,6 +4,7 @@ Release 0.7 (unreleased changes)
 
   NEW FEATURES
 
+   HAMA-736: Add one pass graph generator (tjungblut)
    HAMA-722: Messaging queue should construct sender and receiver queue. (surajsmenon)
    HAMA-721: Added spilling queue with combiner. (surajsmenon)
    HAMA-559: Added spilling queue. (surajsmenon)

Added: hama/trunk/examples/src/main/java/org/apache/hama/examples/util/FastGraphGen.java
URL: http://svn.apache.org/viewvc/hama/trunk/examples/src/main/java/org/apache/hama/examples/util/FastGraphGen.java?rev=1448524&view=auto
==============================================================================
--- hama/trunk/examples/src/main/java/org/apache/hama/examples/util/FastGraphGen.java (added)
+++ hama/trunk/examples/src/main/java/org/apache/hama/examples/util/FastGraphGen.java Thu Feb 21 06:51:57 2013
@@ -0,0 +1,127 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hama.examples.util;
+
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.Random;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hama.HamaConfiguration;
+import org.apache.hama.bsp.BSP;
+import org.apache.hama.bsp.BSPJob;
+import org.apache.hama.bsp.BSPPeer;
+import org.apache.hama.bsp.FileOutputFormat;
+import org.apache.hama.bsp.NullInputFormat;
+import org.apache.hama.bsp.SequenceFileOutputFormat;
+import org.apache.hama.bsp.TextArrayWritable;
+import org.apache.hama.bsp.sync.SyncException;
+import org.apache.hama.examples.CombineExample;
+
+import com.google.common.collect.Sets;
+
+public class FastGraphGen {
+  protected static Log LOG = LogFactory.getLog(FastGraphGen.class);
+
+  private static String SIZE_OF_MATRIX = "size.of.matrix";
+  private static String MAX_EDGES = "max.outlinks";
+
+  public static class SymmetricMatrixGenBSP extends
+      BSP<NullWritable, NullWritable, Text, TextArrayWritable, Text> {
+
+    private Configuration conf;
+    private int sizeN;
+    private int maxOutEdges;
+
+    @Override
+    public void setup(
+        BSPPeer<NullWritable, NullWritable, Text, TextArrayWritable, Text> peer) {
+      this.conf = peer.getConfiguration();
+      sizeN = conf.getInt(SIZE_OF_MATRIX, 10);
+      maxOutEdges = conf.getInt(MAX_EDGES, 1);
+    }
+
+    @Override
+    public void bsp(
+        BSPPeer<NullWritable, NullWritable, Text, TextArrayWritable, Text> peer)
+        throws IOException, SyncException, InterruptedException {
+      int interval = sizeN / peer.getNumPeers();
+      int startID = peer.getPeerIndex() * interval;
+      int endID;
+      if (peer.getPeerIndex() == peer.getNumPeers() - 1) {
+        endID = sizeN;
+      } else {
+        endID = startID + interval;
+      }
+
+      Random r = new Random();
+      for (int i = startID; i < endID; i++) {
+        HashSet<Integer> set = Sets.newHashSet();
+        for (int j = 0; j < maxOutEdges; j++) {
+          set.add(r.nextInt(sizeN));
+        }
+        TextArrayWritable textArrayWritable = new TextArrayWritable();
+        Text[] arr = new Text[set.size()];
+        int index = 0;
+        for (int x : set) {
+          arr[index++] = new Text(x + "");
+        }
+        textArrayWritable.set(arr);
+        peer.write(new Text(i + ""), textArrayWritable);
+      }
+
+    }
+  }
+
+  public static void main(String[] args) throws InterruptedException,
+      IOException, ClassNotFoundException {
+    if (args.length < 4) {
+      System.out
+          .println("Usage: <size n> <max out-edges> <output path> <number of tasks>");
+      System.exit(1);
+    }
+
+    // BSP job configuration
+    HamaConfiguration conf = new HamaConfiguration();
+
+    conf.setInt(SIZE_OF_MATRIX, Integer.parseInt(args[0]));
+    conf.setInt(MAX_EDGES, Integer.parseInt(args[1]));
+
+    BSPJob bsp = new BSPJob(conf, CombineExample.class);
+    // Set the job name
+    bsp.setJobName("Random Fast Matrix Generator");
+    bsp.setBspClass(SymmetricMatrixGenBSP.class);
+    bsp.setInputFormat(NullInputFormat.class);
+    bsp.setOutputKeyClass(Text.class);
+    bsp.setOutputValueClass(TextArrayWritable.class);
+    bsp.setOutputFormat(SequenceFileOutputFormat.class);
+    FileOutputFormat.setOutputPath(bsp, new Path(args[2]));
+    bsp.setNumBspTask(Integer.parseInt(args[3]));
+
+    long startTime = System.currentTimeMillis();
+    if (bsp.waitForCompletion(true)) {
+      System.out.println("Job Finished in "
+          + (System.currentTimeMillis() - startTime) / 1000.0 + " seconds");
+    }
+  }
+}

Modified: hama/trunk/examples/src/main/java/org/apache/hama/examples/util/Generator.java
URL: http://svn.apache.org/viewvc/hama/trunk/examples/src/main/java/org/apache/hama/examples/util/Generator.java?rev=1448524&r1=1448523&r2=1448524&view=diff
==============================================================================
--- hama/trunk/examples/src/main/java/org/apache/hama/examples/util/Generator.java (original)
+++ hama/trunk/examples/src/main/java/org/apache/hama/examples/util/Generator.java Thu Feb 21 06:51:57 2013
@@ -24,6 +24,8 @@ public class Generator {
       System.out.println("Valid command names are:");
       System.out
           .println("  symmetric: Generate random symmetric matrix, which can be used as a input of graph examples.");
+      System.out
+          .println("  fastgen: Generate random matrix, which can be used as a input of graph examples and is faster than symmetric.");
       System.out.println("  square: Generate random square matrix.");
       System.exit(1);
     }
@@ -33,6 +35,8 @@ public class Generator {
 
     if (args[0].equals("symmetric")) {
       SymmetricMatrixGen.main(newArgs);
+    } else if (args[0].equals("fastgen")) {
+      FastGraphGen.main(newArgs);
     } else if (args[0].equals("square")) {
       System.out.println("Not implemented yet.");
       // SquareMatrixGen.main(newArgs);

Added: hama/trunk/examples/src/test/java/org/apache/hama/examples/FastGraphGenTest.java
URL: http://svn.apache.org/viewvc/hama/trunk/examples/src/test/java/org/apache/hama/examples/FastGraphGenTest.java?rev=1448524&view=auto
==============================================================================
--- hama/trunk/examples/src/test/java/org/apache/hama/examples/FastGraphGenTest.java (added)
+++ hama/trunk/examples/src/test/java/org/apache/hama/examples/FastGraphGenTest.java Thu Feb 21 06:51:57 2013
@@ -0,0 +1,67 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hama.examples;
+
+import junit.framework.TestCase;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.SequenceFile;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.Writable;
+import org.apache.hama.bsp.TextArrayWritable;
+import org.apache.hama.examples.util.FastGraphGen;
+import org.junit.Test;
+
+public class FastGraphGenTest extends TestCase {
+  protected static Log LOG = LogFactory.getLog(FastGraphGenTest.class);
+  private static String TEST_OUTPUT = "/tmp/test";
+
+  @Test
+  public void testGraphGenerator() throws Exception {
+    Configuration conf = new Configuration();
+
+    FastGraphGen.main(new String[] { "20", "10", TEST_OUTPUT, "3" });
+    FileSystem fs = FileSystem.get(conf);
+
+    FileStatus[] globStatus = fs.globStatus(new Path(TEST_OUTPUT + "/part-*"));
+    for (FileStatus fts : globStatus) {
+      SequenceFile.Reader reader = new SequenceFile.Reader(fs, fts.getPath(),
+          conf);
+      Text key = new Text();
+      TextArrayWritable value = new TextArrayWritable();
+
+      while (reader.next(key, value)) {
+        Writable[] writables = value.get();
+        assertTrue(writables.length <= 10);
+        for (Writable t : writables) {
+          int outlinkId = Integer.parseInt(t.toString());
+          assertTrue(outlinkId <= 20);
+          assertTrue(outlinkId >= 0);
+        }
+      }
+      reader.close();
+    }
+
+    fs.delete(new Path(TEST_OUTPUT), true);
+  }
+}