import org.apache.giraph.io.VertexReader; //导入依赖的package包/类
@Override
public VertexReader<I, V, E> createVertexReader(InputSplit split,
    TaskAttemptContext context) throws IOException {
  HiveVertexReader<I, V, E> reader = new HiveVertexReader<I, V, E>();
  reader.setTableSchema(getTableSchema());

  RecordReader<WritableComparable, HiveReadableRecord> baseReader;
  try {
    baseReader = hiveInputFormat.createRecordReader(split, context);
  } catch (InterruptedException e) {
    throw new IOException("Could not create vertex reader", e);
  }

  reader.setHiveRecordReader(baseReader);
  return new VertexReaderWrapper<I, V, E>(reader);
}
 

import org.apache.giraph.io.VertexReader; //导入依赖的package包/类
@Override
public final VertexReader<I, V, E>
createVertexReader(final InputSplit split,
                   final TaskAttemptContext context)
  throws IOException {
  try {
    HCatalogVertexReader reader = createVertexReader();
    reader.initialize(hCatInputFormat.
        createVertexRecordReader(split, context));
    return reader;
  } catch (InterruptedException e) {
    throw new IllegalStateException(
        "createVertexReader: " +
            "Interrupted creating reader.", e);
  }
}
 

import org.apache.giraph.io.VertexReader; //导入依赖的package包/类
@Override
public VertexReader<I, V, E> createVertexReader(InputSplit split,
    TaskAttemptContext context) throws IOException {
  HiveVertexReader<I, V, E> reader = new HiveVertexReader<I, V, E>();
  reader.setTableSchema(hiveInputFormat.getTableSchema(getConf()));

  RecordReader<WritableComparable, HiveReadableRecord> baseReader;
  try {
    baseReader = hiveInputFormat.createRecordReader(split, context);
  } catch (InterruptedException e) {
    throw new IOException("Could not create vertex reader", e);
  }

  reader.setHiveRecordReader(baseReader);
  return new VertexReaderWrapper<I, V, E>(reader);
}
 

import org.apache.giraph.io.VertexReader; //导入依赖的package包/类
@Override
public VertexReader createVertexReader(final InputSplit split, final TaskAttemptContext context) throws IOException {
    try {
        final GiraphVertexReader reader = new GiraphVertexReader();
        reader.initialize(split, context);
        return reader;
    } catch (final InterruptedException e) {
        throw new IOException(e.getMessage(), e);
    }
}
 

import org.apache.giraph.io.VertexReader; //导入依赖的package包/类
/**
 * Creates a new VertexReader
 * @param split the split to be read
 * @param context the information about the task
 * @return VertexReader for LinkRank
 * @throws IOException
 */
public VertexReader<Text, DoubleWritable, NullWritable>
createVertexReader(InputSplit split,
                   TaskAttemptContext context) throws IOException {

  return new NutchTableEdgeVertexReader(split, context);

}
 

import org.apache.giraph.io.VertexReader; //导入依赖的package包/类
/**
 * Creates a new VertexReader
 * @param split the split to be read
 * @param context the information about the task
 * @return VertexReader for LinkRank
 * @throws java.io.IOException
 */
public VertexReader<Text, DoubleWritable, NullWritable>
createVertexReader(InputSplit split,
                   TaskAttemptContext context) throws IOException {

  return new NutchTableEdgeVertexReader(split, context);

}
 

import org.apache.giraph.io.VertexReader; //导入依赖的package包/类
public VertexReader<Text, Text, Text>
createVertexReader(InputSplit split, TaskAttemptContext context)
    throws IOException {
  try {

    return new AccumuloEdgeVertexReader(
        accumuloInputFormat.createRecordReader(split, context)) {
    };
  } catch (InterruptedException e) {
    throw new IOException(e);
  }

}
 

import org.apache.giraph.io.VertexReader; //导入依赖的package包/类
@Override
public VertexReader<LongWritable, DoubleWritable,
FloatWritable> createVertexReader(InputSplit split,
  TaskAttemptContext context)
  throws IOException {
  return new SimpleVertexReader();
}
 

import org.apache.giraph.io.VertexReader; //导入依赖的package包/类
@Override
public VertexReader<LongWritable, DoubleWritable,
FloatWritable> createVertexReader(InputSplit split,
  TaskAttemptContext context)
  throws IOException {
  return new SimplePageRankVertexReader();
}
 

import org.apache.giraph.io.VertexReader; //导入依赖的package包/类
@Override
public VertexReader<I, V, E> createVertexReader(InputSplit split,
    TaskAttemptContext context) throws IOException {
  final VertexReader<I, V, E> vertexReader =
      originalInputFormat.createVertexReader(split,
          HadoopUtils.makeTaskAttemptContext(getConf(), context));
  return new WrappedVertexReader<I, V, E>(vertexReader, getConf());
}
 

import org.apache.giraph.io.VertexReader; //导入依赖的package包/类
/**
 * Constructor
 *
 * @param baseVertexReader VertexReader to delegate all the methods to
 * @param conf Configuration
 */
public WrappedVertexReader(VertexReader<I, V, E> baseVertexReader,
    ImmutableClassesGiraphConfiguration<I, V, E> conf) {
  this.baseVertexReader = baseVertexReader;
  super.setConf(conf);
  baseVertexReader.setConf(conf);
}
 

import org.apache.giraph.io.VertexReader; //导入依赖的package包/类
@Override
public VertexReader<Text, RyaTypeWritable, RyaStatementWritable> createVertexReader(InputSplit split, TaskAttemptContext context) throws IOException {
     return new RyaVertexReader((RyaStatementRecordReader) ryaInputFormat.createRecordReader(split, context),
            rdfTableLayout, tripleContext,
            context.getConfiguration());
}
 

import org.apache.giraph.io.VertexReader; //导入依赖的package包/类
@Override
public VertexReader<LongWritable, IntWritable, FloatWritable>
createVertexReader(InputSplit split, TaskAttemptContext context)
  throws IOException {
  return new SimpleSuperstepVertexReader();
}
 

import org.apache.giraph.io.VertexReader; //导入依赖的package包/类
@Override
public VertexReader<I, V, E> createVertexReader(InputSplit inputSplit,
    TaskAttemptContext context) throws IOException {
  return new InMemoryVertexReader();
}
 

import org.apache.giraph.io.VertexReader; //导入依赖的package包/类
@Override
public VertexReader<IntWritable, FloatWritable, NullWritable>
createVertexReader(InputSplit split,
    TaskAttemptContext context) throws IOException {
  return new PseudoRandomVertexReader();
}
 

import org.apache.giraph.io.VertexReader; //导入依赖的package包/类
@Override
public VertexReader<LongWritable, DoubleWritable, DoubleWritable>
createVertexReader(InputSplit split,
    TaskAttemptContext context) throws IOException {
  return new WattsStrogatzVertexReader();
}
 

import org.apache.giraph.io.VertexReader; //导入依赖的package包/类
@Override
public VertexReader<I, V, E> createVertexReader(InputSplit split,
    TaskAttemptContext context) throws IOException {
  return new SequenceFileVertexReader<I, V, E, X>(
      sequenceFileInputFormat.createRecordReader(split, context));
}
 

import org.apache.giraph.io.VertexReader; //导入依赖的package包/类
@Override
public VertexReader<LongWritable, DoubleWritable, DoubleWritable>
createVertexReader(InputSplit split, TaskAttemptContext context)
  throws IOException {
  return new PseudoRandomVertexReader();
}
 

import org.apache.giraph.io.VertexReader; //导入依赖的package包/类
/**
 * Read vertices from input split.  If testing, the user may request a
 * maximum number of vertices to be read from an input split.
 *
 * @param inputSplit Input split to process with vertex reader
 * @param graphState Current graph state
 * @return Vertices and edges loaded from this input split
 * @throws IOException
 * @throws InterruptedException
 */
@Override
protected VertexEdgeCount readInputSplit(
    InputSplit inputSplit,
    GraphState<I, V, E, M> graphState)
  throws IOException, InterruptedException {
  VertexInputFormat<I, V, E> vertexInputFormat =
      configuration.createVertexInputFormat();
  VertexReader<I, V, E> vertexReader =
      vertexInputFormat.createVertexReader(inputSplit, context);
  vertexReader.setConf(
      (ImmutableClassesGiraphConfiguration<I, V, E, Writable>) configuration);
  vertexReader.initialize(inputSplit, context);
  long inputSplitVerticesLoaded = 0;
  long edgesSinceLastUpdate = 0;
  long inputSplitEdgesLoaded = 0;
  while (vertexReader.nextVertex()) {
    Vertex<I, V, E, M> readerVertex =
        (Vertex<I, V, E, M>) vertexReader.getCurrentVertex();
    if (readerVertex.getId() == null) {
      throw new IllegalArgumentException(
          "readInputSplit: Vertex reader returned a vertex " +
              "without an id!  - " + readerVertex);
    }
    if (readerVertex.getValue() == null) {
      readerVertex.setValue(configuration.createVertexValue());
    }
    readerVertex.setConf(configuration);
    readerVertex.setGraphState(graphState);

    PartitionOwner partitionOwner =
        bspServiceWorker.getVertexPartitionOwner(readerVertex.getId());
    graphState.getWorkerClientRequestProcessor().sendVertexRequest(
        partitionOwner, readerVertex);
    context.progress(); // do this before potential data transfer
    ++inputSplitVerticesLoaded;
    edgesSinceLastUpdate += readerVertex.getNumEdges();

    // Update status every VERTICES_UPDATE_PERIOD vertices
    if (inputSplitVerticesLoaded % VERTICES_UPDATE_PERIOD == 0) {
      totalVerticesMeter.mark(VERTICES_UPDATE_PERIOD);
      totalEdgesMeter.mark(edgesSinceLastUpdate);
      inputSplitEdgesLoaded += edgesSinceLastUpdate;
      edgesSinceLastUpdate = 0;

      LoggerUtils.setStatusAndLog(
          context, LOG, Level.INFO,
          "readVertexInputSplit: Loaded " +
              totalVerticesMeter.count() + " vertices at " +
              totalVerticesMeter.meanRate() + " vertices/sec " +
              totalEdgesMeter.count() + " edges at " +
              totalEdgesMeter.meanRate() + " edges/sec " +
              MemoryUtils.getRuntimeMemoryStats());
    }

    // For sampling, or to limit outlier input splits, the number of
    // records per input split can be limited
    if (inputSplitMaxVertices > 0 &&
        inputSplitVerticesLoaded >= inputSplitMaxVertices) {
      if (LOG.isInfoEnabled()) {
        LOG.info("readInputSplit: Leaving the input " +
            "split early, reached maximum vertices " +
            inputSplitVerticesLoaded);
      }
      break;
    }
  }
  vertexReader.close();
  return new VertexEdgeCount(inputSplitVerticesLoaded,
      inputSplitEdgesLoaded + edgesSinceLastUpdate);
}