You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hama.apache.org by ed...@apache.org on 2011/05/30 12:27:54 UTC
svn commit: r1129118 - in /incubator/hama/trunk: ./
src/examples/org/apache/hama/examples/
src/examples/org/apache/hama/examples/sssp/ src/java/org/apache/hama/bsp/
Author: edwardyoon
Date: Mon May 30 10:27:53 2011
New Revision: 1129118
URL: http://svn.apache.org/viewvc?rev=1129118&view=rev
Log:
Add shortest path finding example
Added:
incubator/hama/trunk/src/examples/org/apache/hama/examples/sssp/
incubator/hama/trunk/src/examples/org/apache/hama/examples/sssp/ShortestPathVertex.java
incubator/hama/trunk/src/examples/org/apache/hama/examples/sssp/ShortestPaths.java
incubator/hama/trunk/src/examples/org/apache/hama/examples/sssp/ShortestPathsGraphLoader.java
incubator/hama/trunk/src/examples/org/apache/hama/examples/sssp/ShortestPathsIOUtils.java
incubator/hama/trunk/src/java/org/apache/hama/bsp/BooleanMessage.java
incubator/hama/trunk/src/java/org/apache/hama/bsp/IntegerMessage.java
Modified:
incubator/hama/trunk/CHANGES.txt
incubator/hama/trunk/src/examples/org/apache/hama/examples/ExampleDriver.java
Modified: incubator/hama/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/CHANGES.txt?rev=1129118&r1=1129117&r2=1129118&view=diff
==============================================================================
--- incubator/hama/trunk/CHANGES.txt (original)
+++ incubator/hama/trunk/CHANGES.txt Mon May 30 10:27:53 2011
@@ -4,6 +4,7 @@ Release 0.3 - Unreleased
NEW FEATURES
+ HAMA-359: Add shortest path finding example (Thomas Jungblut via edwardyoon)
HAMA-374: Add LocalBSPRunner (Thomas Jungblut via edwardyoon)
HAMA-380: Add BSPMessageBundle to reduce RPC overhead (Miklos Erdelyi via edwardyoon)
Modified: incubator/hama/trunk/src/examples/org/apache/hama/examples/ExampleDriver.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/src/examples/org/apache/hama/examples/ExampleDriver.java?rev=1129118&r1=1129117&r2=1129118&view=diff
==============================================================================
--- incubator/hama/trunk/src/examples/org/apache/hama/examples/ExampleDriver.java (original)
+++ incubator/hama/trunk/src/examples/org/apache/hama/examples/ExampleDriver.java Mon May 30 10:27:53 2011
@@ -20,16 +20,18 @@
package org.apache.hama.examples;
import org.apache.hadoop.util.ProgramDriver;
+import org.apache.hama.examples.sssp.ShortestPaths;
public class ExampleDriver {
-
+
public static void main(String[] args) {
ProgramDriver pgd = new ProgramDriver();
try {
pgd.addClass("pi", PiEstimator.class, "Pi Estimator");
pgd.addClass("bench", RandBench.class, "Random Communication Benchmark");
pgd.addClass("test", SerializePrinting.class, "Serialize Printing Test");
-
+ pgd.addClass("sssp", ShortestPaths.class, "Single Source Shortest Path");
+
pgd.driver(args);
} catch (Throwable e) {
e.printStackTrace();
Added: incubator/hama/trunk/src/examples/org/apache/hama/examples/sssp/ShortestPathVertex.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/src/examples/org/apache/hama/examples/sssp/ShortestPathVertex.java?rev=1129118&view=auto
==============================================================================
--- incubator/hama/trunk/src/examples/org/apache/hama/examples/sssp/ShortestPathVertex.java (added)
+++ incubator/hama/trunk/src/examples/org/apache/hama/examples/sssp/ShortestPathVertex.java Mon May 30 10:27:53 2011
@@ -0,0 +1,92 @@
+/**
+ * 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.sssp;
+
+public final class ShortestPathVertex {
+
+ private final int id;
+ private final String name;
+ private final int weight;
+ private Integer cost;
+
+ public ShortestPathVertex(int weight, String name) {
+ super();
+ this.id = name.hashCode();
+ this.weight = weight;
+ this.name = name;
+ }
+
+ public ShortestPathVertex(int weight, String name, Integer cost) {
+ super();
+ this.id = name.hashCode();
+ this.weight = weight;
+ this.name = name;
+ this.cost = cost;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public Integer getCost() {
+ return cost;
+ }
+
+ public void setCost(Integer cost) {
+ this.cost = cost;
+ }
+
+ public int getId() {
+ return id;
+ }
+
+ public int getWeight() {
+ return weight;
+ }
+
+ @Override
+ public String toString() {
+ return name;
+ }
+
+ @Override
+ public int hashCode() {
+ final int prime = 31;
+ int result = 1;
+ result = prime * result + ((name == null) ? 0 : name.hashCode());
+ return result;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj)
+ return true;
+ if (obj == null)
+ return false;
+ if (getClass() != obj.getClass())
+ return false;
+ ShortestPathVertex other = (ShortestPathVertex) obj;
+ if (name == null) {
+ if (other.name != null)
+ return false;
+ } else if (!name.equals(other.name))
+ return false;
+ return true;
+ }
+
+}
Added: incubator/hama/trunk/src/examples/org/apache/hama/examples/sssp/ShortestPaths.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/src/examples/org/apache/hama/examples/sssp/ShortestPaths.java?rev=1129118&view=auto
==============================================================================
--- incubator/hama/trunk/src/examples/org/apache/hama/examples/sssp/ShortestPaths.java (added)
+++ incubator/hama/trunk/src/examples/org/apache/hama/examples/sssp/ShortestPaths.java Mon May 30 10:27:53 2011
@@ -0,0 +1,269 @@
+/**
+ * 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.sssp;
+
+import static org.apache.hama.examples.sssp.ShortestPathsIOUtils.mapAdjacencyList;
+import static org.apache.hama.examples.sssp.ShortestPathsIOUtils.partition;
+import static org.apache.hama.examples.sssp.ShortestPathsIOUtils.printOutput;
+import static org.apache.hama.examples.sssp.ShortestPathsIOUtils.saveVertexMap;
+
+import java.io.IOException;
+import java.util.Deque;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hama.HamaConfiguration;
+import org.apache.hama.bsp.BSP;
+import org.apache.hama.bsp.BSPJob;
+import org.apache.hama.bsp.BSPJobClient;
+import org.apache.hama.bsp.BSPPeerProtocol;
+import org.apache.hama.bsp.BooleanMessage;
+import org.apache.hama.bsp.ClusterStatus;
+import org.apache.hama.bsp.IntegerMessage;
+import org.apache.hama.examples.RandBench;
+import org.apache.zookeeper.KeeperException;
+
+public class ShortestPaths extends BSP {
+
+ public static final Log LOG = LogFactory.getLog(ShortestPaths.class);
+
+ public static final String BSP_PEERS = "bsp.peers";
+ public static final String SHORTEST_PATHS_START_VERTEX_ID = "shortest.paths.start.vertex.id";
+ public static final String PARTED = "parted";
+ public static final String IN_PATH = "in.path.";
+ public static final String OUT_PATH = "out.path";
+ public static final String NAME_VALUE_SEPARATOR = ":";
+ public static final String MASTER_TASK = "master.groom";
+
+ private Configuration conf;
+ private Map<ShortestPathVertex, List<ShortestPathVertex>> adjacencyList = new HashMap<ShortestPathVertex, List<ShortestPathVertex>>();
+ private Map<String, ShortestPathVertex> vertexLookupMap = new HashMap<String, ShortestPathVertex>();
+ private String[] peerNames;
+
+ @Override
+ public void bsp(BSPPeerProtocol peer) throws IOException, KeeperException,
+ InterruptedException {
+ LOG.info("Mapping graph into ram...");
+ // map our input into ram
+ mapAdjacencyList(conf, peer, adjacencyList, vertexLookupMap);
+ LOG.info("Finished! Starting graph initialization...");
+ // parse the configuration to get the peerNames
+ parsePeerNames(conf);
+ // get our master groom
+ String master = conf.get(MASTER_TASK);
+
+ // initial message bypass
+ ShortestPathVertex v = vertexLookupMap.get(conf
+ .get(SHORTEST_PATHS_START_VERTEX_ID));
+ if (v != null) {
+ v.setCost(0);
+ sendMessageToNeighbors(peer, v);
+ }
+
+ LOG.info("Finished! Starting main loop...");
+ boolean updated = true;
+ while (updated) {
+ int updatesMade = 0;
+ peer.sync();
+
+ IntegerMessage msg = null;
+ Deque<ShortestPathVertex> updatedQueue = new LinkedList<ShortestPathVertex>();
+ while ((msg = (IntegerMessage) peer.getCurrentMessage()) != null) {
+ ShortestPathVertex vertex = vertexLookupMap.get(msg.getTag());
+ // check if we need an distance update
+ if (vertex.getCost() > msg.getData()) {
+ updatesMade++;
+ updatedQueue.add(vertex);
+ vertex.setCost(msg.getData());
+ }
+ }
+ // synchonize with all grooms if there were updates
+ updated = broadcastUpdatesMade(peer, master, updatesMade);
+ // send updates to the adjacents of the updated vertices
+ for (ShortestPathVertex vertex : updatedQueue) {
+ sendMessageToNeighbors(peer, vertex);
+ }
+ }
+ LOG.info("Finished!");
+ // finished, finally save our map to DFS.
+ saveVertexMap(conf, peer, adjacencyList);
+ }
+
+ /**
+ * Parses the peer names to fix inconsistency in bsp peer names from context.
+ *
+ * @param conf
+ */
+ private void parsePeerNames(Configuration conf) {
+ peerNames = conf.get(BSP_PEERS).split(";");
+ }
+
+ /**
+ * This method broadcasts to a master groom how many updates were made. He
+ * simply sums them up and sends a message back to the grooms if sum is
+ * greater than zero.
+ *
+ * @param peer The peer we got through the BSP method.
+ * @param master The assigned master groom name.
+ * @param updates How many updates were made?
+ * @return True if we need another iteration, False if no updates can be made
+ * anymore.
+ * @throws IOException
+ * @throws KeeperException
+ * @throws InterruptedException
+ */
+ private boolean broadcastUpdatesMade(BSPPeerProtocol peer, String master,
+ int updates) throws IOException, KeeperException, InterruptedException {
+ peer.send(master, new IntegerMessage(peer.getPeerName(), updates));
+ peer.sync();
+ if (peer.getPeerName().equals(master)) {
+ int count = 0;
+ IntegerMessage message;
+ while ((message = (IntegerMessage) peer.getCurrentMessage()) != null) {
+ count += message.getData();
+ LOG.info("Received " + message.getData() + " updates from "
+ + message.getTag() + " in SuperStep " + peer.getSuperstepCount());
+ }
+
+ for (String name : peer.getAllPeerNames()) {
+ peer.send(name, new BooleanMessage("", count > 0 ? true : false));
+ }
+ }
+
+ peer.sync();
+ BooleanMessage message = (BooleanMessage) peer.getCurrentMessage();
+ return message.getData();
+ }
+
+ /**
+ * This method takes advantage of our partitioning: it uses the vertexID
+ * (simply hash of the name) to determine the host where the message belongs
+ * to. <br/>
+ * It sends the current cost to the adjacent vertex + the edge weight. If cost
+ * will be infinity we just going to send infinity, because summing the weight
+ * will cause an integer overflow resulting in negative weights.
+ *
+ * @param peer The peer we got through the BSP method.
+ * @param id The vertex to all adjacent vertices the new cost has to be send.
+ * @throws IOException
+ */
+ private void sendMessageToNeighbors(BSPPeerProtocol peer,
+ ShortestPathVertex id) throws IOException {
+ List<ShortestPathVertex> outgoingEdges = adjacencyList.get(id);
+ for (ShortestPathVertex adjacent : outgoingEdges) {
+ int mod = Math.abs((adjacent.getId() % peer.getAllPeerNames().length));
+ peer.send(peerNames[mod],
+ new IntegerMessage(adjacent.getName(),
+ id.getCost() == Integer.MAX_VALUE ? id.getCost() : id.getCost()
+ + adjacent.getWeight()));
+ }
+ }
+
+ @Override
+ public void setConf(Configuration conf) {
+ this.conf = conf;
+ }
+
+ @Override
+ public Configuration getConf() {
+ return conf;
+ }
+
+ public static void printUsage() {
+ System.out.println("Single Source Shortest Path Example:");
+ System.out
+ .println("<Startvertex name> <optional: output path> <optional: path to own adjacency list sequencefile>");
+ }
+
+ public static void main(String[] args) throws IOException,
+ InterruptedException, ClassNotFoundException {
+
+ printUsage();
+
+ // BSP job configuration
+ HamaConfiguration conf = new HamaConfiguration();
+ conf.set(SHORTEST_PATHS_START_VERTEX_ID, "Frankfurt");
+ System.out.println("Setting default start vertex to \"Frankfurt\"!");
+ conf.set(OUT_PATH, "sssp/output");
+ boolean skipPartitioning = false;
+ Path adjacencyListPath = null;
+
+ if (args.length > 0) {
+ conf.set(SHORTEST_PATHS_START_VERTEX_ID, args[0]);
+ System.out.println("Setting start vertex to " + args[0] + "!");
+
+ if (args.length > 1) {
+ conf.set(OUT_PATH, args[1]);
+ System.out.println("Using new output folder: " + args[1]);
+ }
+
+ if (args.length > 2) {
+ adjacencyListPath = new Path(args[2]);
+ }
+
+ if (args.length > 3) {
+ skipPartitioning = Boolean.valueOf(args[3]);
+ }
+
+ }
+
+ Map<ShortestPathVertex, List<ShortestPathVertex>> adjacencyList = null;
+ if (adjacencyListPath == null)
+ adjacencyList = ShortestPathsGraphLoader.loadGraph();
+
+ BSPJob bsp = new BSPJob(conf, RandBench.class);
+ // Set the job name
+ bsp.setJobName("Single Source Shortest Path");
+ bsp.setBspClass(ShortestPaths.class);
+
+ // Set the task size as a number of GroomServer
+ BSPJobClient jobClient = new BSPJobClient(conf);
+ ClusterStatus cluster = jobClient.getClusterStatus(true);
+ StringBuilder sb = new StringBuilder();
+ for (String peerName : cluster.getActiveGroomNames().values()) {
+ conf.set(MASTER_TASK, peerName);
+ sb.append(peerName);
+ sb.append(";");
+ }
+ LOG.info("Master is: " + conf.get(MASTER_TASK));
+ conf.set(BSP_PEERS, sb.toString());
+ LOG.info("Starting data partitioning...");
+ if (adjacencyList == null)
+ conf = partition(adjacencyListPath, conf, skipPartitioning);
+ else
+ conf = partition(adjacencyList, conf);
+ LOG.info("Finished!");
+
+ bsp.setNumBspTask(cluster.getGroomServers());
+
+ long startTime = System.currentTimeMillis();
+ bsp.waitForCompletion(true);
+ System.out.println("Job Finished in "
+ + (double) (System.currentTimeMillis() - startTime) / 1000.0
+ + " seconds");
+ printOutput(FileSystem.get(conf), conf);
+ }
+
+}
Added: incubator/hama/trunk/src/examples/org/apache/hama/examples/sssp/ShortestPathsGraphLoader.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/src/examples/org/apache/hama/examples/sssp/ShortestPathsGraphLoader.java?rev=1129118&view=auto
==============================================================================
--- incubator/hama/trunk/src/examples/org/apache/hama/examples/sssp/ShortestPathsGraphLoader.java (added)
+++ incubator/hama/trunk/src/examples/org/apache/hama/examples/sssp/ShortestPathsGraphLoader.java Mon May 30 10:27:53 2011
@@ -0,0 +1,94 @@
+/**
+ * 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.sssp;
+
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+
+public class ShortestPathsGraphLoader {
+
+ static Map<ShortestPathVertex, List<ShortestPathVertex>> loadGraph() {
+
+ Map<ShortestPathVertex, List<ShortestPathVertex>> adjacencyList = new HashMap<ShortestPathVertex, List<ShortestPathVertex>>();
+ String[] cities = new String[] { "Frankfurt", "Mannheim", "Wuerzburg",
+ "Stuttgart", "Kassel", "Karlsruhe", "Erfurt", "Nuernberg", "Augsburg",
+ "Muenchen" };
+
+ int id = 1;
+ for (String city : cities) {
+ if (city.equals("Frankfurt")) {
+ List<ShortestPathVertex> list = new LinkedList<ShortestPathVertex>();
+ list.add(new ShortestPathVertex(85, "Mannheim"));
+ list.add(new ShortestPathVertex(173, "Kassel"));
+ list.add(new ShortestPathVertex(217, "Wuerzburg"));
+ adjacencyList.put(new ShortestPathVertex(0, city), list);
+ } else if (city.equals("Stuttgart")) {
+ List<ShortestPathVertex> list = new LinkedList<ShortestPathVertex>();
+ list.add(new ShortestPathVertex(183, "Nuernberg"));
+ adjacencyList.put(new ShortestPathVertex(0, city), list);
+ } else if (city.equals("Kassel")) {
+ List<ShortestPathVertex> list = new LinkedList<ShortestPathVertex>();
+ list.add(new ShortestPathVertex(502, "Muenchen"));
+ list.add(new ShortestPathVertex(173, "Frankfurt"));
+ adjacencyList.put(new ShortestPathVertex(0, city), list);
+ } else if (city.equals("Erfurt")) {
+ List<ShortestPathVertex> list = new LinkedList<ShortestPathVertex>();
+ list.add(new ShortestPathVertex(186, "Wuerzburg"));
+ adjacencyList.put(new ShortestPathVertex(0, city), list);
+ } else if (city.equals("Wuerzburg")) {
+ List<ShortestPathVertex> list = new LinkedList<ShortestPathVertex>();
+ list.add(new ShortestPathVertex(217, "Frankfurt"));
+ list.add(new ShortestPathVertex(168, "Erfurt"));
+ list.add(new ShortestPathVertex(103, "Nuernberg"));
+ adjacencyList.put(new ShortestPathVertex(0, city), list);
+ } else if (city.equals("Mannheim")) {
+ List<ShortestPathVertex> list = new LinkedList<ShortestPathVertex>();
+ list.add(new ShortestPathVertex(80, "Karlsruhe"));
+ list.add(new ShortestPathVertex(85, "Frankfurt"));
+ adjacencyList.put(new ShortestPathVertex(0, city), list);
+ } else if (city.equals("Karlsruhe")) {
+ List<ShortestPathVertex> list = new LinkedList<ShortestPathVertex>();
+ list.add(new ShortestPathVertex(250, "Augsburg"));
+ list.add(new ShortestPathVertex(80, "Mannheim"));
+ adjacencyList.put(new ShortestPathVertex(0, city), list);
+ } else if (city.equals("Augsburg")) {
+ List<ShortestPathVertex> list = new LinkedList<ShortestPathVertex>();
+ list.add(new ShortestPathVertex(250, "Karlsruhe"));
+ list.add(new ShortestPathVertex(84, "Muenchen"));
+ adjacencyList.put(new ShortestPathVertex(0, city), list);
+ } else if (city.equals("Nuernberg")) {
+ List<ShortestPathVertex> list = new LinkedList<ShortestPathVertex>();
+ list.add(new ShortestPathVertex(183, "Stuttgart"));
+ list.add(new ShortestPathVertex(167, "Muenchen"));
+ list.add(new ShortestPathVertex(103, "Wuerzburg"));
+ adjacencyList.put(new ShortestPathVertex(0, city), list);
+ } else if (city.equals("Muenchen")) {
+ List<ShortestPathVertex> list = new LinkedList<ShortestPathVertex>();
+ list.add(new ShortestPathVertex(167, "Nuernberg"));
+ list.add(new ShortestPathVertex(173, "Kassel"));
+ list.add(new ShortestPathVertex(84, "Augsburg"));
+ adjacencyList.put(new ShortestPathVertex(0, city), list);
+ }
+ id++;
+ }
+ return adjacencyList;
+ }
+
+}
Added: incubator/hama/trunk/src/examples/org/apache/hama/examples/sssp/ShortestPathsIOUtils.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/src/examples/org/apache/hama/examples/sssp/ShortestPathsIOUtils.java?rev=1129118&view=auto
==============================================================================
--- incubator/hama/trunk/src/examples/org/apache/hama/examples/sssp/ShortestPathsIOUtils.java (added)
+++ incubator/hama/trunk/src/examples/org/apache/hama/examples/sssp/ShortestPathsIOUtils.java Mon May 30 10:27:53 2011
@@ -0,0 +1,274 @@
+/**
+ * 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.sssp;
+
+import static org.apache.hama.examples.sssp.ShortestPaths.BSP_PEERS;
+import static org.apache.hama.examples.sssp.ShortestPaths.IN_PATH;
+import static org.apache.hama.examples.sssp.ShortestPaths.NAME_VALUE_SEPARATOR;
+import static org.apache.hama.examples.sssp.ShortestPaths.OUT_PATH;
+import static org.apache.hama.examples.sssp.ShortestPaths.PARTED;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+
+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.IntWritable;
+import org.apache.hadoop.io.SequenceFile;
+import org.apache.hadoop.io.Text;
+import org.apache.hama.HamaConfiguration;
+import org.apache.hama.bsp.BSPPeerProtocol;
+
+class ShortestPathsIOUtils {
+
+ /**
+ * When finished we just writing a sequencefile of the vertex name and the
+ * cost.
+ *
+ * @param peer The peer we got through the BSP method.
+ * @param adjacencyList
+ * @throws IOException
+ */
+ static void saveVertexMap(Configuration conf, BSPPeerProtocol peer,
+ Map<ShortestPathVertex, List<ShortestPathVertex>> adjacencyList)
+ throws IOException {
+ Path outPath = new Path(conf.get(OUT_PATH) + Path.SEPARATOR
+ + peer.getPeerName().split(":")[0]);
+ FileSystem fs = FileSystem.get(conf);
+ fs.delete(outPath, true);
+ final SequenceFile.Writer out = SequenceFile.createWriter(fs, conf,
+ outPath, Text.class, IntWritable.class);
+ for (ShortestPathVertex vertex : adjacencyList.keySet()) {
+ out.append(new Text(vertex.getName()), new IntWritable(vertex.getCost()));
+ }
+ out.close();
+ }
+
+ /**
+ * Just a reader of the vertexMap in DFS. Output going to STDOUT.
+ *
+ * @param fs
+ * @param conf
+ * @throws IOException
+ */
+ static void printOutput(FileSystem fs, Configuration conf) throws IOException {
+ System.out.println("-------------------- RESULTS --------------------");
+ FileStatus[] stati = fs.listStatus(new Path(conf.get(OUT_PATH)));
+ for (FileStatus status : stati) {
+ Path path = status.getPath();
+ SequenceFile.Reader reader = new SequenceFile.Reader(fs, path, conf);
+ Text key = new Text();
+ IntWritable value = new IntWritable();
+ while (reader.next(key, value)) {
+ System.out.println(key.toString() + " | " + value.get());
+ }
+ reader.close();
+ }
+ }
+
+ /**
+ *
+ * The adjacencylist contains two text fields on each line. The key component
+ * is the name of a vertex, the value is a ":" separated Text field that
+ * contains the name of the adjacent vertex leftmost and the weight on the
+ * rightmost side.
+ *
+ * <PRE>
+ * K V <br/>
+ * Vertex[Text] AdjacentVertex : Weight [Text]
+ * </PRE>
+ *
+ * @param adjacencyList
+ * @param vertexLookupMap
+ */
+ static void mapAdjacencyList(Configuration conf, BSPPeerProtocol peer,
+ Map<ShortestPathVertex, List<ShortestPathVertex>> adjacencyList,
+ Map<String, ShortestPathVertex> vertexLookupMap)
+ throws FileNotFoundException, IOException {
+ FileSystem fs = FileSystem.get(conf);
+ Path p = new Path(conf.get(IN_PATH
+ + peer.getPeerName().split(NAME_VALUE_SEPARATOR)[0]));
+ SequenceFile.Reader reader = new SequenceFile.Reader(fs, p, conf);
+ Text key = new Text(); // name of the vertex
+ Text value = new Text(); // name of the adjacent vertex : weight
+ while (reader.next(key, value)) {
+ // a key vertex has weight 0 to itself
+ ShortestPathVertex keyVertex = new ShortestPathVertex(0, key.toString(),
+ Integer.MAX_VALUE);
+ String[] nameWeight = value.toString().split(NAME_VALUE_SEPARATOR);
+ if (!adjacencyList.containsKey(keyVertex)) {
+ LinkedList<ShortestPathVertex> list = new LinkedList<ShortestPathVertex>();
+ list.add(new ShortestPathVertex(Integer.valueOf(nameWeight[1]),
+ nameWeight[0], Integer.MAX_VALUE));
+ adjacencyList.put(keyVertex, list);
+ vertexLookupMap.put(keyVertex.getName(), keyVertex);
+ } else {
+ adjacencyList.get(keyVertex).add(
+ new ShortestPathVertex(Integer.valueOf(nameWeight[1]),
+ nameWeight[0], Integer.MAX_VALUE));
+ }
+ }
+ reader.close();
+ }
+
+ /**
+ * Partitioning for in memory adjacency lists.
+ *
+ * @param adjacencyList
+ * @param status
+ * @param conf
+ * @return
+ * @throws IOException
+ */
+ static HamaConfiguration partition(
+ Map<ShortestPathVertex, List<ShortestPathVertex>> adjacencyList,
+ HamaConfiguration conf) throws IOException {
+
+ String[] groomNames = conf.get(BSP_PEERS).split(";");
+
+ int sizeOfCluster = groomNames.length;
+
+ // 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);
+ Path fileToPartition = new Path(conf.get(OUT_PATH));
+ for (String entry : groomNames) {
+ partPaths.add(new Path(fileToPartition.getParent().toString()
+ + Path.SEPARATOR + PARTED + Path.SEPARATOR
+ + entry.split(NAME_VALUE_SEPARATOR)[0]));
+ }
+ // create a seq writer for that
+ for (Path p : partPaths) {
+ // System.out.println(p.toString());
+ fs.delete(p, true);
+ writers.add(SequenceFile
+ .createWriter(fs, conf, p, Text.class, Text.class));
+ }
+
+ for (Entry<ShortestPathVertex, List<ShortestPathVertex>> entry : adjacencyList
+ .entrySet()) {
+ // a key vertex has weight 0 to itself
+ ShortestPathVertex keyVertex = entry.getKey();
+ // just mod the id
+ int mod = Math.abs(keyVertex.getId() % sizeOfCluster);
+ // append it to the right sequenceFile
+ for (ShortestPathVertex value : entry.getValue())
+ writers.get(mod)
+ .append(
+ new Text(keyVertex.getName()),
+ new Text(value.getName() + NAME_VALUE_SEPARATOR
+ + value.getWeight()));
+ }
+
+ for (SequenceFile.Writer w : writers)
+ w.close();
+
+ for (Path p : partPaths) {
+ conf.set(IN_PATH + p.getName(), p.toString());
+ }
+ return conf;
+ }
+
+ /**
+ * Partitioning for sequencefile partitioned adjacency lists.
+ *
+ * The adjacencylist contains two text fields on each line. The key component
+ * is the name of a vertex, the value is a ":" separated Text field that
+ * contains the name of the adjacent vertex leftmost and the weight on the
+ * rightmost side.
+ *
+ * <PRE>
+ * K V <br/>
+ * Vertex[Text] AdjacentVertex : Weight [Text]
+ * </PRE>
+ *
+ * @param fileToPartition
+ * @param status
+ * @param conf
+ * @return
+ * @throws IOException
+ */
+ static HamaConfiguration partition(Path fileToPartition,
+ HamaConfiguration conf, boolean skipPartitioning) throws IOException {
+
+ String[] groomNames = conf.get(BSP_PEERS).split(";");
+ int sizeOfCluster = groomNames.length;
+
+ // 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 (String entry : groomNames) {
+ partPaths.add(new Path(fileToPartition.getParent().toString()
+ + Path.SEPARATOR + PARTED + Path.SEPARATOR
+ + entry.split(NAME_VALUE_SEPARATOR)[0]));
+ }
+
+ if (!skipPartitioning) {
+
+ // create a seq writer for that
+ for (Path p : partPaths) {
+ // System.out.println(p.toString());
+ fs.delete(p, true);
+ writers.add(SequenceFile.createWriter(fs, conf, p, Text.class,
+ Text.class));
+ }
+
+ // parse our file
+ if (!fs.exists(fileToPartition))
+ throw new FileNotFoundException("File " + fileToPartition
+ + " wasn't found!");
+
+ SequenceFile.Reader reader = new SequenceFile.Reader(fs, fileToPartition,
+ conf);
+ Text key = new Text(); // name of the vertex
+ Text value = new Text(); // name of the adjacent vertex : weight
+
+ while (reader.next(key, value)) {
+ // a key vertex has weight 0 to itself
+ ShortestPathVertex keyVertex = new ShortestPathVertex(0, key.toString());
+ // just mod the id
+ int mod = Math.abs(keyVertex.getId() % sizeOfCluster);
+ // append it to the right sequenceFile
+ writers.get(mod).append(new Text(keyVertex.getName()), new Text(value));
+ }
+
+ reader.close();
+
+ for (SequenceFile.Writer w : writers)
+ w.close();
+ }
+
+ for (Path p : partPaths) {
+ conf.set(IN_PATH + p.getName(), p.toString());
+ }
+
+ return conf;
+ }
+
+}
Added: incubator/hama/trunk/src/java/org/apache/hama/bsp/BooleanMessage.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/src/java/org/apache/hama/bsp/BooleanMessage.java?rev=1129118&view=auto
==============================================================================
--- incubator/hama/trunk/src/java/org/apache/hama/bsp/BooleanMessage.java (added)
+++ incubator/hama/trunk/src/java/org/apache/hama/bsp/BooleanMessage.java Mon May 30 10:27:53 2011
@@ -0,0 +1,60 @@
+/**
+ * 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.bsp;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+public class BooleanMessage extends BSPMessage {
+
+ String tag;
+ boolean data;
+
+ public BooleanMessage() {
+ super();
+ }
+
+ public BooleanMessage(String tag, boolean data) {
+ super();
+ this.tag = tag;
+ this.data = data;
+ }
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ out.writeUTF(tag);
+ out.writeBoolean(data);
+ }
+
+ @Override
+ public void readFields(DataInput in) throws IOException {
+ tag = in.readUTF();
+ data = in.readBoolean();
+ }
+
+ @Override
+ public String getTag() {
+ return tag;
+ }
+
+ @Override
+ public Boolean getData() {
+ return data;
+ }
+}
\ No newline at end of file
Added: incubator/hama/trunk/src/java/org/apache/hama/bsp/IntegerMessage.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/src/java/org/apache/hama/bsp/IntegerMessage.java?rev=1129118&view=auto
==============================================================================
--- incubator/hama/trunk/src/java/org/apache/hama/bsp/IntegerMessage.java (added)
+++ incubator/hama/trunk/src/java/org/apache/hama/bsp/IntegerMessage.java Mon May 30 10:27:53 2011
@@ -0,0 +1,61 @@
+/**
+ * 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.bsp;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+public class IntegerMessage extends BSPMessage {
+
+ String tag;
+ int data;
+
+ public IntegerMessage() {
+ super();
+ }
+
+ public IntegerMessage(String tag, int data) {
+ super();
+ this.tag = tag;
+ this.data = data;
+ }
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ out.writeUTF(tag);
+ out.writeInt(data);
+ }
+
+ @Override
+ public void readFields(DataInput in) throws IOException {
+ tag = in.readUTF();
+ data = in.readInt();
+ }
+
+ @Override
+ public String getTag() {
+ return tag;
+ }
+
+ @Override
+ public Integer getData() {
+ return data;
+ }
+
+}
\ No newline at end of file