code: public <T> DataSet<T> groupReduceOnEdges(EdgesFunction<K, EV, T> edgesFunction,
  			EdgeDirection direction, TypeInformation<T> typeInfo) throws IllegalArgumentException {
  
  		switch (direction) {
  			case IN:
  				return edges.map(new ProjectVertexIdMap<>(1)).name("Vertex ID")
  						.withForwardedFields("f1->f0")
  						.groupBy(0).reduceGroup(new ApplyGroupReduceFunction<>(edgesFunction))
  							.name("GroupReduce on in-edges").returns(typeInfo);
  			case OUT:
  				return edges.map(new ProjectVertexIdMap<>(0)).name("Vertex ID")
  						.withForwardedFields("f0")
  						.groupBy(0).reduceGroup(new ApplyGroupReduceFunction<>(edgesFunction))
  							.name("GroupReduce on out-edges").returns(typeInfo);
  			case ALL:
  				return edges.flatMap(new EmitOneEdgePerNode<>()).name("Emit edge")
  						.groupBy(0).reduceGroup(new ApplyGroupReduceFunction<>(edgesFunction))
  							.name("GroupReduce on in- and out-edges").returns(typeInfo);
  			default:
  				throw new IllegalArgumentException("Illegal edge direction");
  		}
  	}
docs: Groups by vertex and computes a GroupReduce transformation over the edge values of each vertex.
  The edgesFunction applied on the edges only has access to the vertex id (not the vertex value)
  of the grouping vertex.
  
  <p>For each vertex, the edgesFunction can iterate over all edges of this vertex
  with the specified direction, and emit any number of output elements, including none.
  
  @param edgesFunction the group reduce function to apply to the neighboring edges of each vertex.
  @param direction the edge direction (in-, out-, all-).
  @param <T> the output type
  @param typeInfo the explicit return type.
  @return a DataSet containing elements of type T
  @throws IllegalArgumentException