import org.apache.tinkerpop.gremlin.process.computer.KeyValue; //导入依赖的package包/类
@Override
public KeyValue next() {
    try {
        if (this.available) {
            this.available = false;
            return new KeyValue<>(this.key.get(), this.value.get());
        } else {
            while (true) {
                if (this.readers.isEmpty())
                    throw new NoSuchElementException();
                if (this.readers.peek().next(this.key, this.value)) {
                    return new KeyValue<>(this.key.get(), this.value.get());
                } else
                    this.readers.remove().close();
            }
        }
    } catch (final IOException e) {
        throw new IllegalStateException(e.getMessage(), e);
    }
}
 

import org.apache.tinkerpop.gremlin.process.computer.KeyValue; //导入依赖的package包/类
@Override
public <K, V> Iterator<KeyValue<K, V>> writeMemoryRDD(final Configuration configuration, final String memoryKey, JavaPairRDD<K, V> memoryRDD) {
    final org.apache.hadoop.conf.Configuration hadoopConfiguration = ConfUtil.makeHadoopConfiguration(configuration);
    final String outputLocation = hadoopConfiguration.get(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION);
    if (null != outputLocation) {
        // map back to a Hadoop stream for output
        memoryRDD.mapToPair(keyValue -> new Tuple2<>(new ObjectWritable<>(keyValue._1()), new ObjectWritable<>(keyValue._2())))
                .saveAsNewAPIHadoopFile(Constants.getMemoryLocation(outputLocation, memoryKey),
                        ObjectWritable.class,
                        ObjectWritable.class,
                        SequenceFileOutputFormat.class, hadoopConfiguration);
        try {
            return (Iterator) new ObjectWritableIterator(hadoopConfiguration, new Path(Constants.getMemoryLocation(outputLocation, memoryKey)));
        } catch (final IOException e) {
            throw new IllegalStateException(e.getMessage(), e);
        }
    }
    return Collections.emptyIterator();
}
 

import org.apache.tinkerpop.gremlin.process.computer.KeyValue; //导入依赖的package包/类
@Override
public <K, V> Iterator<KeyValue<K, V>> head(final String location, final String memoryKey, final Class readerClass, final int totalLines) {
    final Configuration configuration = new BaseConfiguration();
    configuration.setProperty(Constants.GREMLIN_HADOOP_INPUT_LOCATION, location);
    configuration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_READER, readerClass.getCanonicalName());
    try {
        if (InputRDD.class.isAssignableFrom(readerClass)) {
            return IteratorUtils.map(((InputRDD) readerClass.getConstructor().newInstance()).readMemoryRDD(configuration, memoryKey, new JavaSparkContext(Spark.getContext())).take(totalLines).iterator(), tuple -> new KeyValue(tuple._1(), tuple._2()));
        } else if (InputFormat.class.isAssignableFrom(readerClass)) {
            return IteratorUtils.map(new InputFormatRDD().readMemoryRDD(configuration, memoryKey, new JavaSparkContext(Spark.getContext())).take(totalLines).iterator(), tuple -> new KeyValue(tuple._1(), tuple._2()));
        }
    } catch (final Exception e) {
        throw new IllegalArgumentException(e.getMessage(), e);
    }
    throw new IllegalArgumentException("The provided parserClass must be an " + InputFormat.class.getCanonicalName() + " or an " + InputRDD.class.getCanonicalName() + ": " + readerClass.getCanonicalName());
}
 

import org.apache.tinkerpop.gremlin.process.computer.KeyValue; //导入依赖的package包/类
protected void complete(final MapReduce<?, ?, OK, OV, ?> mapReduce) {
    if (mapReduce.getReduceKeySort().isPresent()) {
        final Comparator<OK> comparator = mapReduce.getReduceKeySort().get();
        final List<KeyValue<OK, OV>> list = new ArrayList<>(this.reduceQueue);
        Collections.sort(list, Comparator.comparing(KeyValue::getKey, comparator));
        this.reduceQueue.clear();
        this.reduceQueue.addAll(list);
    }
}
 

import org.apache.tinkerpop.gremlin.process.computer.KeyValue; //导入依赖的package包/类
@Override
public void emit(K key, V value) {
    if (this.doReduce)
        this.reduceMap.computeIfAbsent(key, k -> new ConcurrentLinkedQueue<>()).add(value);
    else
        this.mapQueue.add(new KeyValue<>(key, value));
}
 

import org.apache.tinkerpop.gremlin.process.computer.KeyValue; //导入依赖的package包/类
@Override
public Map<Serializable, Set<String>> generateFinalResult(Iterator<KeyValue<Serializable, Set<String>>> keyValues) {
    if (this.persistentProperties.containsKey(CLUSTER_SIZE)) {
        long clusterSize = (long) persistentProperties.get(CLUSTER_SIZE);
        keyValues = IteratorUtils.filter(keyValues, pair -> Long.valueOf(pair.getValue().size()).equals(clusterSize));
    }
    final Map<Serializable, Set<String>> clusterPopulation = Utility.keyValuesToMap(keyValues);
    clusterPopulation.remove(NullObject.instance());
    return clusterPopulation;
}
 

import org.apache.tinkerpop.gremlin.process.computer.KeyValue; //导入依赖的package包/类
@Override
public Map<Serializable, Long> generateFinalResult(Iterator<KeyValue<Serializable, Long>> keyValues) {
    if (this.persistentProperties.containsKey(CLUSTER_SIZE)) {
        long clusterSize = (long) persistentProperties.get(CLUSTER_SIZE);
        keyValues = IteratorUtils.filter(keyValues, pair -> pair.getValue().equals(clusterSize));
    }
    final Map<Serializable, Long> clusterPopulation = Utility.keyValuesToMap(keyValues);
    clusterPopulation.remove(NullObject.instance());
    return clusterPopulation;
}
 

import org.apache.tinkerpop.gremlin.process.computer.KeyValue; //导入依赖的package包/类
@Override
public <K, V> Iterator<KeyValue<K, V>> head(final String location, final String memoryKey, final Class readerClass, final int totalLines) {
    if (!readerClass.equals(SequenceFileInputFormat.class))
        throw new IllegalArgumentException("Only " + SequenceFileInputFormat.class.getCanonicalName() + " memories are supported");
    final Configuration configuration = new Configuration();
    try {
        return IteratorUtils.limit((Iterator) new ObjectWritableIterator(configuration, new Path(Constants.getMemoryLocation(location, memoryKey))), totalLines);
    } catch (final IOException e) {
        throw new IllegalStateException(e.getMessage(), e);
    }
}
 

import org.apache.tinkerpop.gremlin.process.computer.KeyValue; //导入依赖的package包/类
@Override
public <K, V> Iterator<KeyValue<K, V>> writeMemoryRDD(final Configuration configuration, final String memoryKey, final JavaPairRDD<K, V> memoryRDD) {
    if (!configuration.getBoolean(Constants.GREMLIN_SPARK_PERSIST_CONTEXT, false))
        LOGGER.warn("The SparkContext should be persisted in order for the RDD to persist across jobs. To do so, set " + Constants.GREMLIN_SPARK_PERSIST_CONTEXT + " to true");
    if (!configuration.containsKey(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION))
        throw new IllegalArgumentException("There is no provided " + Constants.GREMLIN_HADOOP_OUTPUT_LOCATION + " to write the persisted RDD to");
    final String memoryRDDName = Constants.getMemoryLocation(configuration.getString(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION), memoryKey);
    Spark.removeRDD(memoryRDDName);
    memoryRDD.setName(memoryRDDName).persist(StorageLevel.fromString(configuration.getString(Constants.GREMLIN_SPARK_PERSIST_STORAGE_LEVEL, "MEMORY_ONLY")));
    Spark.refresh(); // necessary to do really fast so the Spark GC doesn't clear out the RDD
    return IteratorUtils.map(memoryRDD.collect().iterator(), tuple -> new KeyValue<>(tuple._1(), tuple._2()));
}
 

import org.apache.tinkerpop.gremlin.process.computer.KeyValue; //导入依赖的package包/类
@Override
   public void emit(K key, V value) {
if (doReduce)
    reduceMap.computeIfAbsent(key, k -> new ConcurrentLinkedQueue<>()).add(value);
else
    mapQueue.add(new KeyValue<>(key, value));
   }
 

import org.apache.tinkerpop.gremlin.process.computer.KeyValue; //导入依赖的package包/类
protected void complete(MapReduce<?, ?, OK, OV, ?> mapReduce) {
if (mapReduce.getReduceKeySort().isPresent()) {
    Comparator<OK> comparator = mapReduce.getReduceKeySort().get();
    List<KeyValue<OK, OV>> list = new ArrayList<>(reduceQueue);
    Collections.sort(list, Comparator.comparing(KeyValue::getKey, comparator));
    reduceQueue.clear();
    reduceQueue.addAll(list);
}
   }
 

import org.apache.tinkerpop.gremlin.process.computer.KeyValue; //导入依赖的package包/类
@Override
public void emit(final OK key, final OV value) {
    this.reduceQueue.add(new KeyValue<>(key, value));
}
 

import org.apache.tinkerpop.gremlin.process.computer.KeyValue; //导入依赖的package包/类
@Override
public Map<Serializable, T> generateFinalResult(Iterator<KeyValue<Serializable, T>> iterator) {
    return Utility.keyValuesToMap(iterator);
}
 

import org.apache.tinkerpop.gremlin.process.computer.KeyValue; //导入依赖的package包/类
@Override
public Map<Serializable, Set<String>> generateFinalResult(Iterator<KeyValue<Serializable, Set<String>>> keyValues) {
    final Map<Serializable, Set<String>> clusterPopulation = Utility.keyValuesToMap(keyValues);
    clusterPopulation.remove(NullObject.instance());
    return clusterPopulation;
}
 

import org.apache.tinkerpop.gremlin.process.computer.KeyValue; //导入依赖的package包/类
@Override
public Iterator<KeyValue<Object, Double>> generateFinalResult(final Iterator<KeyValue<Object, Double>> keyValues) {
    return keyValues;
}
 

import org.apache.tinkerpop.gremlin.process.computer.KeyValue; //导入依赖的package包/类
@Override
public Map<Serializable, Long> generateFinalResult(final Iterator<KeyValue<Serializable, Long>> keyValues) {
    final Map<Serializable, Long> clusterPopulation = new HashMap<>();
    keyValues.forEachRemaining(pair -> clusterPopulation.put(pair.getKey(), pair.getValue()));
    return clusterPopulation;
}
 

import org.apache.tinkerpop.gremlin.process.computer.KeyValue; //导入依赖的package包/类
@Override
public Integer generateFinalResult(final Iterator<KeyValue<NullObject, Integer>> keyValues) {
    return keyValues.next().getValue();
}
 

import org.apache.tinkerpop.gremlin.process.computer.KeyValue; //导入依赖的package包/类
@Override
public Long generateFinalResult(final Iterator<KeyValue<String, Long>> keyValues) {
    return keyValues.next().getValue();
}
 

import org.apache.tinkerpop.gremlin.process.computer.KeyValue; //导入依赖的package包/类
@Override
public Iterator<KeyValue<Object, Long>> generateFinalResult(final Iterator<KeyValue<Object, Long>> keyValues) {
    return keyValues;
}
 

import org.apache.tinkerpop.gremlin.process.computer.KeyValue; //导入依赖的package包/类
@Override
   public void emit(OK key, OV value) {
reduceQueue.add(new KeyValue<>(key, value));
   }