giraph-examples/src/main/java/org/apache/giraph/examples/PageRankWithKryoSimpleWritable.java

/*
 * 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.giraph.examples;

import com.google.common.collect.Lists;
import org.apache.giraph.aggregators.BasicAggregator;
import org.apache.giraph.aggregators.LongSumAggregator;
import org.apache.giraph.edge.Edge;
import org.apache.giraph.edge.EdgeFactory;
import org.apache.giraph.graph.BasicComputation;
import org.apache.giraph.graph.Vertex;
import org.apache.giraph.io.VertexReader;
import org.apache.giraph.io.formats.GeneratedVertexInputFormat;
import org.apache.giraph.master.DefaultMasterCompute;
import org.apache.giraph.worker.WorkerContext;
import org.apache.giraph.writable.kryo.KryoSimpleWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.log4j.Logger;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import org.apache.giraph.examples.PageRankWithKryoSimpleWritable.VertexValue;
import org.apache.giraph.examples.PageRankWithKryoSimpleWritable.MessageValue;
import org.apache.giraph.examples.PageRankWithKryoSimpleWritable.EdgeValue;

/**
 * Copy of SimplePageRank, modified to test vertex/edge and
 * message values that derives from KryoSimpleWritable.
 */
@Algorithm(
        name = "Page rank"
)
public class PageRankWithKryoSimpleWritable extends
        BasicComputation<LongWritable, VertexValue,
        EdgeValue, MessageValue> {
  /** Number of supersteps for this test */
  public static final int MAX_SUPERSTEPS = 30;
  /** Number of supersteps for this static  3;
  /** Logger */
  private static final Logger LOG =
          Logger.getLogger(PageRankWithKryoSimpleWritable.class);
  /** Sum aggregator name */
  private static String SUM_AGG = "sum";
  /** Min aggregator name */
  private static String MIN_AGG = "min";
  /** Max aggregator name */
  private static String MAX_AGG = "max";

  @Override
  public void compute(
          Vertex<LongWritable, VertexValue,
          EdgeValue> vertex,
          Iterable<MessageValue> messages) throws IOException {
    if (getSuperstep() >= 1) {
      double sum = 0;
      for (MessageValue message : messages) {
        sum += message.get();
      }
      Double value = (0.15f / getTotalNumVertices()) + 0.85f * sum;
      VertexValue vertexValue = new VertexValue(value);
      vertex.setValue(vertexValue);
      aggregate(MAX_AGG, vertexValue);
      aggregate(MIN_AGG, vertexValue);
      aggregate(SUM_AGG, new LongWritable(1));
      LOG.info(vertex.getId() + ": PageRank=" + vertexValue +
              " max=" + getAggregatedValue(MAX_AGG) +
              " min=" + getAggregatedValue(MIN_AGG));
    }

    if (getSuperstep() < MAX_SUPERSTEPS) {
      long edges = vertex.getNumEdges();
      sendMessageToAllEdges(vertex,
          new MessageValue(vertex.getValue().get() / edges));
    } else {
      vertex.voteToHalt();
    }
  }

  /**
   * Worker context used with {@link PageRankWithKryoSimpleWritable}.
   */
  public static class PageRankWithKryoWorkerContext extends
          WorkerContext {
    /** Final max value for verification for local jobs */
    private static double FINAL_MAX;
    /** Final min value for verification for local jobs */
    private static double FINAL_MIN;
    /** Final sum value for verification for local jobs */
    private static long FINAL_SUM;

    public static double getFinalMax() {
      return FINAL_MAX;
    }

    public static double getFinalMin() {
      return FINAL_MIN;
    }

    public static long getFinalSum() {
      return FINAL_SUM;
    }

    @Override
    public void preApplication()
            throws InstantiationException, IllegalAccessException {
    }

    @Override
    public void postApplication() {
      FINAL_SUM = this.<LongWritable>getAggregatedValue(SUM_AGG).get();
      FINAL_MAX = this.<VertexValue>getAggregatedValue(MAX_AGG).get();
      FINAL_MIN = this.<VertexValue>getAggregatedValue(MIN_AGG).get();

      LOG.info("aggregatedNumVertices=" + FINAL_SUM);
      LOG.info("aggregatedMaxPageRank=" + FINAL_MAX);
      LOG.info("aggregatedMinPageRank=" + FINAL_MIN);
    }

    @Override
    public void preSuperstep() {
      if (getSuperstep() >= 3) {
        LOG.info("aggregatedNumVertices=" +
                getAggregatedValue(SUM_AGG) +
                " NumVertices=" + getTotalNumVertices());
        if (this.<LongWritable>getAggregatedValue(SUM_AGG).get() !=
                getTotalNumVertices()) {
          throw new RuntimeException("wrong value of SumAggreg: " +
                  getAggregatedValue(SUM_AGG) + ", should be: " +
                  getTotalNumVertices());
        }
        VertexValue maxPagerank = getAggregatedValue(MAX_AGG);
        LOG.info("aggregatedMaxPageRank=" + maxPagerank.get());
        VertexValue minPagerank = getAggregatedValue(MIN_AGG);
        LOG.info("aggregatedMinPageRank=" + minPagerank.get());
      }
    }

    @Override
    public void postSuperstep() { }
  }

  /**
   * Master compute associated with {@link PageRankWithKryoSimpleWritable}.
   * It registers required aggregators.
   */
  public static class PageRankWithKryoMasterCompute extends
          DefaultMasterCompute {
    @Override
    public void initialize() throws InstantiationException,
            IllegalAccessException {
      registerAggregator(SUM_AGG, LongSumAggregator.class);
      registerPersistentAggregator(MIN_AGG, DoubleMinWrapperAggregator.class);
      registerPersistentAggregator(MAX_AGG, DoubleMaxWrapperAggregator.class);
    }
  }

  /**
   * Simple VertexReader that supports {@link PageRankWithKryoSimpleWritable}
   */
  public static class PageRankWithKryoVertexReader extends
          GeneratedVertexReader<LongWritable, VertexValue, EdgeValue> {
    /** Class logger */
    private static final Logger LOG =
        Logger.getLogger(
          PageRankWithKryoSimpleWritable.PageRankWithKryoVertexReader.class);

    @Override
    public boolean nextVertex() {
      return totalRecords > recordsRead;
    }

    @Override
    public Vertex<LongWritable, VertexValue, EdgeValue>
    getCurrentVertex() throws IOException {
      Vertex<LongWritable, VertexValue, EdgeValue> vertex =
              getConf().createVertex();
      LongWritable vertexId = new LongWritable(
              (inputSplit.getSplitIndex() * totalRecords) + recordsRead);
      VertexValue vertexValue = new VertexValue(vertexId.get() * 10d);
      long targetVertexId =
              (vertexId.get() + 1) %
                      (inputSplit.getNumSplits() * totalRecords);
      float edgeValue = vertexId.get() * 100f;
      List<Edge<LongWritable, EdgeValue>> edges = Lists.newLinkedList();
      edges.add(EdgeFactory.create(new LongWritable(targetVertexId),
              new EdgeValue(edgeValue)));
      vertex.initialize(vertexId, vertexValue, edges);
      ++recordsRead;
      if (LOG.isInfoEnabled()) {
        LOG.info("next: Return vertexId=" + vertex.getId().get() +
          ", vertexValue=" + vertex.getValue() +
          ", targetVertexId=" + targetVertexId + ", edgeValue=" + edgeValue);
      }
      return vertex;
    }
  }

  /**
   *  VertexInputFormat that supports {@link PageRankWithKryoSimpleWritable}
   */
  public static class PageRankWithKryoVertexInputFormat extends
          GeneratedVertexInputFormat<LongWritable, VertexValue, EdgeValue> {
    @Override
    public VertexReader<LongWritable, VertexValue,
            EdgeValue> createVertexReader(InputSplit split,
                                              TaskAttemptContext context)
            throws IOException {
      return new PageRankWithKryoVertexReader();
    }
  }

  /**
   * Creating a custom vertex value class to force kryo to
   * register with a new ID. Please note that a custom
   * class containing a double array is not
   * necessary for the page rank application. It is only
   * used for testing the scenario of kryo encountering an
   * unregistered custom class.
   */
  public static class VertexValue extends KryoSimpleWritable {
    /** Storing the value in an array.
        Double array is an unregistered type
        hence kryo will assign a unique class id */
    private double[] ranks;

    /** Constructor */
    public VertexValue() {
    }

    /**
     * Constructor
     * @param val Vertex value
     */
    public VertexValue(Double val) {
      ranks = new double[1];

      ranks[0] = val;
    }

    /**
     * Get vertex value
     * @return Vertex value
     */
    public Double get() {
      return ranks[0];
    }

    /**
     * Set vertex value.
     * @param val Vertex value
     */
    public void set(Double val) {
      this.ranks[0] = val;
    }
  }

  /**
   * Creating a custom edge value class to force kryo to
   * register with a new ID. Please note that a custom
   * class containing a float is not
   * necessary for the page rank application. It is only
   * used for testing the scenario of kryo encountering an
   * unregistered custom class.
   */
  public static class EdgeValue extends KryoSimpleWritable {
    /** Edge value */
    private Float realValue;

    /** Constructor */
    public EdgeValue() {
    }
    /**
     * Constructor
     * @param val Edge value
     */
    public EdgeValue(Float val) {
      realValue = val;
    }

    /**
     * Get edge value
     * @return Edge value
     */
    public Float get() {
      return realValue;
    }

    /**
     * Set edge value
     * @param val Edge value
     */
    public void set(Float val) {
      this.realValue = val;
    }
  }

  /**
   * Creating a custom message value class to force kryo to
   * register with a new ID. Please note that a custom
   * class containing a double list is not
   * necessary for the page rank application. It is only
   * used for testing the scenario of kryo encountering an
   * unregistered custom class.
   */
  public static class MessageValue extends KryoSimpleWritable {
    /** Storing the message in a list to test the list type */
    private List<Double> msgValue;

    /** Constructor */
    public MessageValue() {
    }

    /**
     * Constructor
     * @param val Message value
     */
    public MessageValue(Double val) {
      msgValue = new ArrayList<>();
      msgValue.add(val);
    }

    /**
     * Get message value
     * @return Message value
     */
    public Double get() {
      return msgValue.get(0);
    }

    /**
     * Set message value
     * @param val Message value
     */
    public void set(Double val) {
      this.msgValue.set(0, val);
    }
  }


  /**
   * Aggregator for getting max double value
   */
  public static class DoubleMaxWrapperAggregator extends
          BasicAggregator<VertexValue> {
    @Override
    public void aggregate(VertexValue value) {
      getAggregatedValue().set(
                      Math.max(getAggregatedValue().get(), value.get()));
    }

    @Override
    public VertexValue createInitialValue() {
      return new VertexValue(Double.NEGATIVE_INFINITY);
    }
  }

  /**
   * Aggregator for getting min double value.
   */
  public static class DoubleMinWrapperAggregator
          extends BasicAggregator<VertexValue> {
    @Override
    public void aggregate(VertexValue value) {
      getAggregatedValue().set(
              Math.min(getAggregatedValue().get(), value.get()));
    }

    @Override
    public VertexValue createInitialValue() {
      return  new VertexValue(Double.MAX_VALUE);
    }
  }

}