anchor.diff:  Implementing MapReduce Emitters
   +++++++++++++++++++++++++++++++
   
  -image:hadoop-logo-notext.png[width=150,float=left] The MapReduce framework in TinkerPop3 is similar to the model popularized by link:http://apache.hadoop.org[Hadoop]. The primary difference is that all Mappers process the vertices of the graph, not an arbitrary key/value pair. A Gremlin OLAP vendor needs to provide implementations for to particular classes: `MapReduce.MapEmitter` and `MapReduce.ReduceEmitter`. TinkerGraph's implementation is provided below which demonstrates the simplicity of the algorithm (especially when the data is all within the same JVM).
  -
  -[source,java]
  -----
  -class TinkerMapEmitter<K, V> implements MapReduce.MapEmitter<K, V> {
  -
  -    public Map<K, Queue<V>> reduceMap = new ConcurrentHashMap<>();
  -    public Queue<Pair<K, V>> mapQueue = new ConcurrentLinkedQueue<>();
  +image:hadoop-logo-notext.png[width=150,float=left] The MapReduce framework in TinkerPop3 is similar to the model popularized by link:http://apache.hadoop.org[Hadoop]. The primary difference is that all Mappers process the vertices of the graph, not an arbitrary key/value pair. However, the vertices' edges can not be accessed -- only their properties. This greatly reduces the amount of data needed to be pushed through the MapReduce engine as any edge information required, can be computed in the VertexProgram.execute() method. Moreover, at this stage, vertices can not be mutated, only their token and property data read. A Gremlin OLAP vendor needs to provide implementations for to particular classes: `MapReduce.MapEmitter` and `MapReduce.ReduceEmitter`. TinkerGraph's implementation is provided below which demonstrates the simplicity of the algorithm (especially when the data is all within the same JVM).
  +
  +[source,java]
  +----
  +public class TinkerMapEmitter<K, V> implements MapReduce.MapEmitter<K, V> {
  +
  +    public Map<K, Queue<V>> reduceMap;
  +    public Queue<KeyValue<K, V>> mapQueue;
       private final boolean doReduce;
   
  -    public TinkerMapEmitter(final boolean doReduce) {  <1>
  +    public TinkerMapEmitter(final boolean doReduce) { <1>
           this.doReduce = doReduce;
  +        if (this.doReduce)
  +            this.reduceMap = new ConcurrentHashMap<>();
  +        else
  +            this.mapQueue = new ConcurrentLinkedQueue<>();
       }
   
       @Override
       public void emit(K key, V value) {
           if (this.doReduce)
  -            MapHelper.concurrentIncr(this.reduceMap, key, value); <2>
  +            this.reduceMap.computeIfAbsent(key, k -> new ConcurrentLinkedQueue<>()).add(value); <2>
           else
  -            this.mapQueue.add(new Pair<>(key, value)); <3>
  +            this.mapQueue.add(new KeyValue<>(key, value)); <3>
  +    }
  +
  +    protected void complete(final MapReduce<K, V, ?, ?, ?> mapReduce) {
  +        if (!this.doReduce && mapReduce.getMapKeySort().isPresent()) {
  +            final Comparator<K> comparator = mapReduce.getMapKeySort().get();
  +            final List<KeyValue<K, V>> list = new ArrayList<>(this.mapQueue);
  +            Collections.sort(list, Comparator.comparing(KeyValue::getKey, comparator));
  +            this.mapQueue.clear();
  +            this.mapQueue.addAll(list);
  +        } else if (mapReduce.getMapKeySort().isPresent()) {
  +            final Comparator<K> comparator = mapReduce.getMapKeySort().get();
  +            final List<Map.Entry<K, Queue<V>>> list = new ArrayList<>();
  +            list.addAll(this.reduceMap.entrySet());
  +            Collections.sort(list, Comparator.comparing(Map.Entry::getKey, comparator));
  +            this.reduceMap = new LinkedHashMap<>();
  +            list.forEach(entry -> this.reduceMap.put(entry.getKey(), entry.getValue()));
  +        }
       }
   }
   ----
   
  -<1> If the MapReduce job has a reduce, then use one data structure (`reduceMap`), else use another (`mapList`). The difference being that a reduction requires a grouping by key and therefore, the `Map<K,Queue<V>>` definition. If no reduction/grouping is required, then a simple `Queue<Pair<K,V>>` can be leveraged.
  +<1> If the MapReduce job has a reduce, then use one data structure (`reduceMap`), else use another (`mapList`). The difference being that a reduction requires a grouping by key and therefore, the `Map<K,Queue<V>>` definition. If no reduction/grouping is required, then a simple `Queue<KeyValue<K,V>>` can be leveraged.
   <2> If reduce is to follow, then increment the Map with a new value for the key. `MapHelper` is a TinkerPop3 class with static methods for adding data to a Map.
  -<3> If no reduce is to follow, then simply append a Pair to the queue.
  -
  -[source,java]
  -----
  -class TinkerReduceEmitter<OK, OV> implements MapReduce.ReduceEmitter<OK, OV> {
  -
  -    public Queue<Pair<OK, OV>> resultList = new ConcurrentLinkedQueue<>();
  +<3> If no reduce is to follow, then simply append a KeyValue to the queue.
  +<4> When the map phase is complete, any map-result sorting required can be executed at this point.
  +
  +[source,java]
  +----
  +public class TinkerReduceEmitter<OK, OV> implements MapReduce.ReduceEmitter<OK, OV> {
  +
  +    protected Queue<KeyValue<OK, OV>> reduceQueue = new ConcurrentLinkedQueue<>();
   
       @Override
       public void emit(final OK key, final OV value) {
  -        this.resultList.add(new Pair<>(key, value));
  +        this.reduceQueue.add(new KeyValue<>(key, value));
  +    }
  +
  +    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);
  +        }
       }
   }
   ----
  
positive.code: The method `MapReduce.reduce()` is defined as:
  
  [source,java]
  public void reduce(final MK key, final Iterator<MV> values, final ReduceEmitter<RK, RV> emitter) { ... }
  
  In other words, for the TinkerGraph implementation, iterate through the entrySet of the `reduceMap` and call the `reduce()` method on each entry. The `reduce()` method can emit key/value pairs which are simply aggregated into a `Queue<Pair<OK,OV>>` in an analogous fashion to `TinkerMapEmitter` when no reduce is to follow. These two emitters are tied together in `TinkerGraphComputer.submit()`.
  
  [source,java]
  ...
  for (final MapReduce mapReduce : this.mapReduces) {
      if (mapReduce.doStage(MapReduce.Stage.MAP)) {
          final TinkerMapEmitter<?, ?> mapEmitter = new TinkerMapEmitter<>(mapReduce.doStage(MapReduce.Stage.REDUCE));
          TinkerHelper.getVertices(this.graph).parallelStream().forEach(vertex -> mapReduce.map(vertex, mapEmitter));
          // no need to run combiners as this is single machine
          if (mapReduce.doStage(MapReduce.Stage.REDUCE)) {
              final TinkerReduceEmitter<?, ?> reduceEmitter = new TinkerReduceEmitter<>();
              mapEmitter.reduceMap.entrySet().parallelStream().forEach(entry -> mapReduce.reduce(entry.getKey(), entry.getValue().iterator(), reduceEmitter));
              mapReduce.addSideEffectToMemory(this.memory, reduceEmitter.resultList.iterator()); <1>
          } else {
              mapReduce.addSideEffectToMemory(this.memory, mapEmitter.mapQueue.iterator()); <2>
          }
      }
  }
  ...
  
  <1> Note that the final results of the reducer are provided to the Memory as specified by the application developer's `MapReduce.addSideEffectToMemory()` implementation.
  <2> If there is no reduce stage, the the map-stage results are inserted into Memory as specified by the application developer's `MapReduce.addSideEffectToMemory()` implementation.
  
  [[io-implementations]]
  IO Implementations
  
negative.code: Implementing Gremlin-Core
  ~~~~~~~~~~~~~~~~~~~~~~~~~
  
  The classes that a vendor should focus on implementing are itemized below. It is a good idea to study the <<tinkergraph-gremlin,TinkerGraph>> (in-memory OLTP and OLAP in `tinkergraph-gremlin`), <<neo4j-gremlin,Neo4jGraph>> (OTLP w/ transactions in `neo4j-gremlin`) and/or <<hadoop-gremlin,HadoopGraph>> (OLAP in `hadoop-gremlin`) implementations for ideas and patterns.
  
  . Online Transactional Processing Graph Systems (*OLTP*)
   .. Structure API: `Graph`, `Element`, `Vertex`, `Edge`, `Property` and `Transaction` (if transactions are supported).
   .. Process API: `TraversalStrategy` instances for optimizing Gremlin traversals to the vendors graph system (i.e. `TinkerGraphStepStrategy`).
  . Online Analytics Processing Graph Systems (*OLAP*)
   .. Everything required of OTLP is required of OLAP (but not vice versa).
   .. GraphComputer API: `GraphComputer`, `Messenger`, `Memory`.
  