code: @Override
  	public EdgeMetrics<K, VV, EV> run(Graph<K, VV, EV> input)
  			throws Exception {
  		super.run(input);
  
  		// s, t, (d(s), d(t))
  		DataSet<Edge<K, Tuple3<EV, LongValue, LongValue>>> edgeDegreePair = input
  			.run(new EdgeDegreePair<K, VV, EV>()
  				.setReduceOnTargetId(reduceOnTargetId)
  				.setParallelism(parallelism));
  
  		// s, d(s), count of (u, v) where deg(u) < deg(v) or (deg(u) == deg(v) and u < v)
  		DataSet<Tuple3<K, LongValue, LongValue>> edgeStats = edgeDegreePair
  			.map(new EdgeStats<>())
  				.setParallelism(parallelism)
  				.name("Edge stats")
  			.groupBy(0)
  			.reduce(new SumEdgeStats<>())
  			.setCombineHint(CombineHint.HASH)
  				.setParallelism(parallelism)
  				.name("Sum edge stats");
  
  		edgeMetricsHelper = new EdgeMetricsHelper<>();
  
  		edgeStats
  			.output(edgeMetricsHelper)
  				.setParallelism(parallelism)
  				.name("Edge metrics");
  
  		return this;
  	}
docs: /*
  Implementation notes:
  
  <p>Use aggregator to replace SumEdgeStats when aggregators are rewritten to use
  a hash-combineable hashed-reduce.