hama-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From mi...@apache.org
Subject svn commit: r1701975 - in /hama/trunk/yarn: pom.xml src/main/java/org/apache/hama/bsp/PageRankonYarn.java src/main/java/org/apache/hama/bsp/YARNGraphJob.java
Date Wed, 09 Sep 2015 10:52:30 GMT
Author: minho
Date: Wed Sep  9 10:52:30 2015
New Revision: 1701975

URL: http://svn.apache.org/r1701975
Log:
Add the GraphJob example, which is PageRank, to work on YARN cluster.

Added:
    hama/trunk/yarn/src/main/java/org/apache/hama/bsp/PageRankonYarn.java
    hama/trunk/yarn/src/main/java/org/apache/hama/bsp/YARNGraphJob.java
Modified:
    hama/trunk/yarn/pom.xml

Modified: hama/trunk/yarn/pom.xml
URL: http://svn.apache.org/viewvc/hama/trunk/yarn/pom.xml?rev=1701975&r1=1701974&r2=1701975&view=diff
==============================================================================
--- hama/trunk/yarn/pom.xml (original)
+++ hama/trunk/yarn/pom.xml Wed Sep  9 10:52:30 2015
@@ -42,6 +42,11 @@
       </exclusions>
     </dependency>
     <dependency>
+      <groupId>org.apache.hama</groupId>
+      <artifactId>hama-graph</artifactId>
+      <version>${project.version}</version>
+    </dependency>
+    <dependency>
       <groupId>commons-logging</groupId>
       <artifactId>commons-logging</artifactId>
     </dependency>

Added: hama/trunk/yarn/src/main/java/org/apache/hama/bsp/PageRankonYarn.java
URL: http://svn.apache.org/viewvc/hama/trunk/yarn/src/main/java/org/apache/hama/bsp/PageRankonYarn.java?rev=1701975&view=auto
==============================================================================
--- hama/trunk/yarn/src/main/java/org/apache/hama/bsp/PageRankonYarn.java (added)
+++ hama/trunk/yarn/src/main/java/org/apache/hama/bsp/PageRankonYarn.java Wed Sep  9 10:52:30
2015
@@ -0,0 +1,217 @@
+/**
+ * Copyright 2007 The Apache Software Foundation
+ *
+ * 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 org.apache.commons.cli.*;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.DoubleWritable;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hama.HamaConfiguration;
+import org.apache.hama.graph.*;
+import org.json.simple.JSONArray;
+import org.json.simple.parser.JSONParser;
+
+import java.io.IOException;
+import java.util.Iterator;
+
+public class PageRankonYarn {
+  public static class PageRankVertex extends
+      Vertex<Text, NullWritable, DoubleWritable> {
+
+    static double DAMPING_FACTOR = 0.85;
+    static double MAXIMUM_CONVERGENCE_ERROR = 0.001;
+
+    @Override
+    public void setup(HamaConfiguration conf) {
+      String val = conf.get("hama.pagerank.alpha");
+      if (val != null) {
+        DAMPING_FACTOR = Double.parseDouble(val);
+      }
+      val = conf.get("hama.graph.max.convergence.error");
+      if (val != null) {
+        MAXIMUM_CONVERGENCE_ERROR = Double.parseDouble(val);
+      }
+
+      // initialize this vertex to 1 / count of global vertices in this graph
+      setValue(new DoubleWritable(1.0 / getTotalNumVertices()));
+    }
+
+    @Override
+    public void compute(Iterable<DoubleWritable> messages) throws IOException {
+      if (this.getSuperstepCount() >= 1) {
+        double sum = 0;
+        for (DoubleWritable msg : messages) {
+          sum += msg.get();
+        }
+        double alpha = (1.0d - DAMPING_FACTOR) / getTotalNumVertices();
+        setValue(new DoubleWritable(alpha + (sum * DAMPING_FACTOR)));
+        aggregate(0, this.getValue());
+      }
+
+      // if we have not reached our global error yet, then proceed.
+      DoubleWritable globalError = getAggregatedValue(0);
+
+      if (globalError != null && this.getSuperstepCount() > 2
+          && MAXIMUM_CONVERGENCE_ERROR > globalError.get()) {
+        voteToHalt();
+      } else {
+        // in each superstep we are going to send a new rank to our neighbours
+        sendMessageToNeighbors(new DoubleWritable(this.getValue().get()
+            / this.getEdges().size()));
+      }
+    }
+  }
+
+  public static class PagerankTextReader extends
+      VertexInputReader<LongWritable, Text, Text, NullWritable, DoubleWritable> {
+
+    @Override
+    public boolean parseVertex(LongWritable key, Text value,
+        Vertex<Text, NullWritable, DoubleWritable> vertex) throws Exception {
+
+      String[] tokenArray = value.toString().split("\t");
+      String vtx = tokenArray[0].trim();
+      String[] edges = tokenArray[1].trim().split(" ");
+
+      vertex.setVertexID(new Text(vtx));
+
+      for (String v : edges) {
+        vertex.addEdge(new Edge<Text, NullWritable>(new Text(v), null));
+      }
+
+      return true;
+    }
+  }
+
+  public static class PagerankJsonReader extends
+      VertexInputReader<LongWritable, Text, Text, NullWritable, DoubleWritable> {
+
+    @SuppressWarnings("unchecked")
+    @Override
+    public boolean parseVertex(LongWritable key, Text value,
+        Vertex<Text, NullWritable, DoubleWritable> vertex) throws Exception {
+      JSONArray jsonArray = (JSONArray) new JSONParser()
+          .parse(value.toString());
+
+      vertex.setVertexID(new Text(jsonArray.get(0).toString()));
+
+      Iterator<JSONArray> iter = ((JSONArray) jsonArray.get(2)).iterator();
+      while (iter.hasNext()) {
+        JSONArray edge = iter.next();
+        vertex.addEdge(new Edge<Text, NullWritable>(new Text(edge.get(0)
+            .toString()), null));
+      }
+
+      return true;
+    }
+  }
+
+  public static YARNGraphJob createJob(String[] args, HamaConfiguration conf,
+      Options opts) throws IOException, ParseException {
+    CommandLine cliParser = new GnuParser().parse(opts, args);
+
+    if (!cliParser.hasOption("i") || !cliParser.hasOption("o")) {
+      System.out
+          .println("No input or output path specified for PageRank, exiting.");
+    }
+
+    YARNGraphJob pageJob = new YARNGraphJob(conf, PageRankonYarn.class);
+    pageJob.setJobName("Pagerank");
+
+    pageJob.setVertexClass(PageRankVertex.class);
+    pageJob.setInputPath(new Path(cliParser.getOptionValue("i")));
+    pageJob.setOutputPath(new Path(cliParser.getOptionValue("o")));
+
+    // set the defaults
+    pageJob.setMaxIteration(30);
+    pageJob.set("hama.pagerank.alpha", "0.85");
+    // reference vertices to itself, because we don't have a dangling node
+    // contribution here
+    pageJob.set("hama.graph.self.ref", "true");
+    pageJob.set("hama.graph.max.convergence.error", "0.001");
+
+    if (cliParser.hasOption("t")) {
+      pageJob.setNumBspTask(Integer.parseInt(cliParser.getOptionValue("t")));
+    }
+
+    // error
+    pageJob.setAggregatorClass(AverageAggregator.class);
+
+    // Vertex reader
+    // According to file type, which is Text or Json,
+    // Vertex reader handle it differently.
+    if (cliParser.hasOption("f")) {
+      if (cliParser.getOptionValue("f").equals("text")) {
+        pageJob.setVertexInputReaderClass(PagerankTextReader.class);
+      } else if (cliParser.getOptionValue("f").equals("json")) {
+        pageJob.setVertexInputReaderClass(PagerankJsonReader.class);
+      } else {
+        System.out.println("File type is not available to run Pagerank... "
+            + "File type set default value, Text.");
+        pageJob.setVertexInputReaderClass(PagerankTextReader.class);
+      }
+    } else {
+      pageJob.setVertexInputReaderClass(PagerankTextReader.class);
+    }
+
+    pageJob.setVertexIDClass(Text.class);
+    pageJob.setVertexValueClass(DoubleWritable.class);
+    pageJob.setEdgeValueClass(NullWritable.class);
+
+    pageJob.setInputFormat(TextInputFormat.class);
+    pageJob.setInputKeyClass(LongWritable.class);
+    pageJob.setInputValueClass(Text.class);
+
+    pageJob.setPartitioner(HashPartitioner.class);
+    pageJob.setOutputFormat(TextOutputFormat.class);
+    pageJob.setOutputKeyClass(Text.class);
+    pageJob.setOutputValueClass(DoubleWritable.class);
+    return pageJob;
+  }
+
+  public static void main(String[] args) throws IOException,
+      InterruptedException, ClassNotFoundException, ParseException {
+    Options opts = new Options();
+    opts.addOption("i", "input_path", true, "The Location of output path.");
+    opts.addOption("o", "output_path", true, "The Location of input path.");
+    opts.addOption("h", "help", false, "Print usage");
+    opts.addOption("t", "task_num", true, "The number of tasks.");
+    opts.addOption("f", "file_type", true, "The file type of input data. Input"
+        + "file format which is \"text\" tab delimiter separated or \"json\"."
+        + "Default value - Text");
+
+    if (args.length < 2) {
+      new HelpFormatter().printHelp("pagerank -i INPUT_PATH -o OUTPUT_PATH "
+          + "[-t NUM_TASKS] [-f FILE_TYPE]", opts);
+      System.exit(-1);
+    }
+
+    HamaConfiguration conf = new HamaConfiguration();
+    YARNGraphJob pageJob = createJob(args, conf, opts);
+
+    long startTime = System.currentTimeMillis();
+    if (pageJob.waitForCompletion(true)) {
+      System.out.println("Job Finished in "
+          + (System.currentTimeMillis() - startTime) / 1000.0 + " seconds");
+    }
+  }
+}

Added: hama/trunk/yarn/src/main/java/org/apache/hama/bsp/YARNGraphJob.java
URL: http://svn.apache.org/viewvc/hama/trunk/yarn/src/main/java/org/apache/hama/bsp/YARNGraphJob.java?rev=1701975&view=auto
==============================================================================
--- hama/trunk/yarn/src/main/java/org/apache/hama/bsp/YARNGraphJob.java (added)
+++ hama/trunk/yarn/src/main/java/org/apache/hama/bsp/YARNGraphJob.java Wed Sep  9 10:52:30
2015
@@ -0,0 +1,196 @@
+/**
+ * Copyright 2007 The Apache Software Foundation
+ *
+ * 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 com.google.common.base.Preconditions;
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.Writable;
+import org.apache.hama.Constants;
+import org.apache.hama.HamaConfiguration;
+import org.apache.hama.bsp.message.MessageManager;
+import org.apache.hama.bsp.message.OutgoingMessageManager;
+import org.apache.hama.bsp.message.queue.MessageQueue;
+import org.apache.hama.graph.*;
+
+import java.io.IOException;
+
+public class YARNGraphJob extends YARNBSPJob {
+  public final static String VERTEX_CLASS_ATTR = "hama.graph.vertex.class";
+  public final static String VERTEX_ID_CLASS_ATTR = "hama.graph.vertex.id.class";
+  public final static String VERTEX_VALUE_CLASS_ATTR = "hama.graph.vertex.value.class";
+  public final static String VERTEX_EDGE_VALUE_CLASS_ATTR = "hama.graph.vertex.edge.value.class";
+
+  public final static String VERTEX_OUTPUT_WRITER_CLASS_ATTR = "hama.graph.vertex.output.writer.class";
+  public final static String AGGREGATOR_CLASS_ATTR = "hama.graph.aggregator.class";
+
+  public YARNGraphJob(HamaConfiguration conf, Class<?> exampleClass)
+      throws IOException {
+    super(conf);
+    conf.setClass(MessageManager.OUTGOING_MESSAGE_MANAGER_CLASS,
+        OutgoingVertexMessageManager.class, OutgoingMessageManager.class);
+    conf.setBoolean(Constants.FORCE_SET_BSP_TASKS, true);
+    conf.setBoolean(Constants.ENABLE_RUNTIME_PARTITIONING, false);
+    conf.setBoolean("hama.use.unsafeserialization", true);
+
+    this.setBspClass(GraphJobRunner.class);
+    this.setJarByClass(exampleClass);
+    this.setVertexIDClass(Text.class);
+    this.setVertexValueClass(IntWritable.class);
+    this.setEdgeValueClass(IntWritable.class);
+    this.setPartitioner(HashPartitioner.class);
+    this.setMessageQueueBehaviour(MessageQueue.PERSISTENT_QUEUE);
+  }
+
+    /**
+     * Set the Vertex class for the job.
+     */
+
+  public void setVertexClass(Class<? extends
+      Vertex<? extends Writable, ? extends Writable, ? extends Writable>> cls)
+      throws IllegalStateException {
+    conf.setClass(VERTEX_CLASS_ATTR, cls, Vertex.class);
+    setInputKeyClass(cls);
+    setInputValueClass(NullWritable.class);
+  }
+
+  /**
+   * Set the Vertex ID class for the job.
+   */
+  public void setVertexIDClass(Class<? extends Writable> cls)
+      throws IllegalStateException {
+    conf.setClass(VERTEX_ID_CLASS_ATTR, cls, Writable.class);
+  }
+
+  /**
+   * Set the Vertex value class for the job.
+   */
+  public void setVertexValueClass(Class<? extends Writable> cls)
+      throws IllegalStateException {
+    conf.setClass(VERTEX_VALUE_CLASS_ATTR, cls, Writable.class);
+  }
+
+  /**
+   * Set the Edge value class for the job.
+   */
+  public void setEdgeValueClass(Class<? extends Writable> cls)
+      throws IllegalStateException {
+    conf.setClass(VERTEX_EDGE_VALUE_CLASS_ATTR, cls, Writable.class);
+  }
+
+  /**
+   * Set the aggregator for the job.
+   */
+  @SuppressWarnings({ "rawtypes", "unchecked" })
+  public void setAggregatorClass(Class<? extends Aggregator> cls) {
+    this.setAggregatorClass(new Class[] { cls });
+  }
+
+  /**
+   * Sets multiple aggregators for the job.
+   */
+  @SuppressWarnings("rawtypes")
+  public void setAggregatorClass(Class<? extends Aggregator>... cls) {
+    String classNames = "";
+    for (Class<? extends Aggregator> cl : cls) {
+      classNames += cl.getName() + ";";
+    }
+    conf.set(AGGREGATOR_CLASS_ATTR, classNames);
+  }
+
+  /**
+   * Sets the input reader for parsing the input to vertices.
+   */
+  public void setVertexInputReaderClass(
+      @SuppressWarnings("rawtypes") Class<? extends VertexInputReader> cls) {
+    ensureState(JobState.DEFINE);
+    conf.setClass(Constants.RUNTIME_PARTITION_RECORDCONVERTER, cls,
+        PartitioningRunner.RecordConverter.class);
+  }
+
+  /**
+   * Sets the output writer for materializing vertices to the output sink. If
+   * not set, the default DefaultVertexOutputWriter will be used.
+   */
+  public void setVertexOutputWriterClass(
+      @SuppressWarnings("rawtypes") Class<? extends VertexOutputWriter> cls) {
+    ensureState(JobState.DEFINE);
+    conf.setClass(VERTEX_OUTPUT_WRITER_CLASS_ATTR, cls,
+        VertexOutputWriter.class);
+  }
+
+  @SuppressWarnings("unchecked")
+  public Class<? extends Vertex<? extends Writable, ? extends Writable, ? extends Writable>>
getVertexClass() {
+    return (Class<? extends Vertex<? extends Writable, ? extends Writable, ? extends
Writable>>) conf
+        .getClass(VERTEX_CLASS_ATTR, Vertex.class);
+  }
+
+  @Override
+  public void setPartitioner(
+      @SuppressWarnings("rawtypes") Class<? extends Partitioner> theClass) {
+    super.setPartitioner(theClass);
+  }
+
+  @Override
+  public void setCombinerClass(
+      Class<? extends Combiner<? extends Writable>> cls) {
+    ensureState(JobState.DEFINE);
+    conf.setClass(Constants.COMBINER_CLASS, cls, Combiner.class);
+  }
+
+  /**
+   * Sets how many iterations the algorithm should perform, -1 for deactivated
+   * is default value.
+   */
+  public void setMaxIteration(int maxIteration) {
+    conf.setInt("hama.graph.max.iteration", maxIteration);
+  }
+
+  @Override
+  public void submit() throws IOException, InterruptedException {
+    Preconditions
+        .checkArgument(this.getConfiguration().get(VERTEX_CLASS_ATTR) != null,
+            "Please provide a vertex class!");
+    Preconditions.checkArgument(
+        this.getConfiguration().get(VERTEX_ID_CLASS_ATTR) != null,
+        "Please provide an vertex ID class!");
+    Preconditions.checkArgument(
+        this.getConfiguration().get(VERTEX_VALUE_CLASS_ATTR) != null,
+        "Please provide an vertex value class, if you don't need one, use NullWritable!");
+    Preconditions.checkArgument(
+        this.getConfiguration().get(VERTEX_EDGE_VALUE_CLASS_ATTR) != null,
+        "Please provide an edge value class, if you don't need one, use NullWritable!");
+
+    Preconditions.checkArgument(
+        this.getConfiguration().get(Constants.RUNTIME_PARTITION_RECORDCONVERTER)
+            != null,
+        "Please provide a converter class for your vertex by using GraphJob#setVertexInputReaderClass!");
+
+    if (this.getConfiguration().get(VERTEX_OUTPUT_WRITER_CLASS_ATTR) == null) {
+      this.setVertexOutputWriterClass(DefaultVertexOutputWriter.class);
+    }
+
+    this.getConfiguration().setClass(MessageManager.RECEIVE_QUEUE_TYPE_CLASS,
+        IncomingVertexMessageManager.class, MessageQueue.class);
+
+    super.submit();
+  }
+}



Mime
View raw message