Page History

  public static class PagerankTextReader extends
      VertexInputReader<LongWritable, Text, Text, NullWritable, DoubleWritable> {

    String lastVertexId = null;
    List<String> adjacents = new LinkedList<String>ArrayList<String>();

    @Override
    public boolean parseVertex(LongWritable key, Text value,
        Vertex<Text, NullWritable, DoubleWritable> vertex) {

      String line = value.toString();
      String[] lineSplit = line.split("\t");
      if (!line.startsWith("#")) {
        if (lastVertexId == null) {
          lastVertexId = lineSplit[0];
        }
        if (lastVertexId.equals(lineSplit[0])) {
          adjacents.add(lineSplit[1]);
        } else {
          vertex.setVertexID(new Text(lastVertexId));
          for (String adjacent : adjacents) {
            vertex.addEdge(new Edge<Text, NullWritable>(new Text(adjacent),
                null));
          }
          adjacents.clear();
          lastVertexId = lineSplit[0];
          adjacents.add(lineSplit[1]);
          return true;
        }
      }
      return false;
    }

  }