code: @Override
  	public DataSet<Result<K>> runInternal(Graph<K, VV, EV> input)
  			throws Exception {
  		// s, t, d(t)
  		DataSet<Edge<K, Tuple2<EV, LongValue>>> neighborDegree = input
  			.run(new EdgeTargetDegree<K, VV, EV>()
  				.setParallelism(parallelism));
  
  		// group span, s, t, d(t)
  		DataSet<Tuple4<IntValue, K, K, IntValue>> groupSpans = neighborDegree
  			.groupBy(0)
  			.sortGroup(1, Order.ASCENDING)
  			.reduceGroup(new GenerateGroupSpans<>(groupSize))
  				.setParallelism(parallelism)
  				.name("Generate group spans");
  
  		// group, s, t, d(t)
  		DataSet<Tuple4<IntValue, K, K, IntValue>> groups = groupSpans
  			.rebalance()
  				.setParallelism(parallelism)
  				.name("Rebalance")
  			.flatMap(new GenerateGroups<>())
  				.setParallelism(parallelism)
  				.name("Generate groups");
  
  		// t, u, d(t)+d(u)
  		DataSet<Tuple3<K, K, IntValue>> twoPaths = groups
  			.groupBy(0, 1)
  			.sortGroup(2, Order.ASCENDING)
  			.reduceGroup(new GenerateGroupPairs<>(groupSize))
  				.name("Generate group pairs");
  
  		// t, u, intersection, union
  		DataSet<Result<K>> scores = twoPaths
  			.groupBy(0, 1)
  			.reduceGroup(new ComputeScores<>(unboundedScores,
  					minimumScoreNumerator, minimumScoreDenominator,
  					maximumScoreNumerator, maximumScoreDenominator))
  				.name("Compute scores");
  
  		if (mirrorResults) {
  			scores = scores
  				.flatMap(new MirrorResult<>())
  					.name("Mirror results");
  		}
  
  		return scores;
  	}
docs: /*
  Implementation notes:
  
  The requirement that "K extends CopyableValue<K>" can be removed when
  Flink has a self-join which performs the skew distribution handled by
  GenerateGroupSpans / GenerateGroups / GenerateGroupPairs.