code: public DataSet<Tuple2<K, EV>> reduceOnEdges(ReduceEdgesFunction<EV> reduceEdgesFunction,
  			EdgeDirection direction) throws IllegalArgumentException {
  
  		switch (direction) {
  			case IN:
  				return edges.map(new ProjectVertexWithEdgeValueMap<>(1))
  						.withForwardedFields("f1->f0")
  							.name("Vertex with in-edges")
  						.groupBy(0).reduce(new ApplyReduceFunction<>(reduceEdgesFunction))
  							.name("Reduce on edges");
  			case OUT:
  				return edges.map(new ProjectVertexWithEdgeValueMap<>(0))
  						.withForwardedFields("f0->f0")
  							.name("Vertex with out-edges")
  						.groupBy(0).reduce(new ApplyReduceFunction<>(reduceEdgesFunction))
  							.name("Reduce on edges");
  			case ALL:
  				return edges.flatMap(new EmitOneVertexWithEdgeValuePerNode<>())
  						.withForwardedFields("f2->f1")
  							.name("Vertex with all edges")
  						.groupBy(0).reduce(new ApplyReduceFunction<>(reduceEdgesFunction))
  							.name("Reduce on edges");
  			default:
  				throw new IllegalArgumentException("Illegal edge direction");
  		}
  	}
docs: Compute a reduce transformation over the edge values of each vertex.
  For each vertex, the transformation consecutively calls a
  {@link ReduceEdgesFunction} until only a single value for each edge remains.
  The {@link ReduceEdgesFunction} combines two edge values into one new value of the same type.
  
  @param reduceEdgesFunction the reduce function to apply to the neighbors of each vertex.
  @param direction the edge direction (in-, out-, all-)
  @return a Dataset of Tuple2, with one tuple per vertex.
  The first field of the Tuple2 is the vertex ID and the second field
  is the aggregate value computed by the provided {@link ReduceEdgesFunction}.
  @throws IllegalArgumentException