docs: @Override
  	public DataSet<Result<K>> runInternal(Graph<K, VV, EV> input)
  			throws Exception {
  		// s, d(s), 1/log(d(s))
  		DataSet<Tuple3<K, LongValue, FloatValue>> inverseLogDegree = input
  			.run(new VertexDegree<K, VV, EV>()
  				.setParallelism(parallelism))
  			.map(new VertexInverseLogDegree<>())
  				.setParallelism(parallelism)
  				.name("Vertex score");
  
  		// s, t, 1/log(d(s))
  		DataSet<Tuple3<K, K, FloatValue>> sourceInverseLogDegree = input
  			.getEdges()
  			.join(inverseLogDegree, JoinHint.REPARTITION_HASH_SECOND)
  			.where(0)
  			.equalTo(0)
  			.projectFirst(0, 1)
  			.<Tuple3<K, K, FloatValue>>projectSecond(2)
  				.setParallelism(parallelism)
  				.name("Edge score");
  
  		// group span, s, t, 1/log(d(s))
  		DataSet<Tuple4<IntValue, K, K, FloatValue>> groupSpans = sourceInverseLogDegree
  			.groupBy(0)
  			.sortGroup(1, Order.ASCENDING)
  			.reduceGroup(new GenerateGroupSpans<>())
  				.setParallelism(parallelism)
  				.name("Generate group spans");
  
  		// group, s, t, 1/log(d(s))
  		DataSet<Tuple4<IntValue, K, K, FloatValue>> groups = groupSpans
  			.rebalance()
  				.setParallelism(parallelism)
  				.name("Rebalance")
  			.flatMap(new GenerateGroups<>())
  				.setParallelism(parallelism)
  				.name("Generate groups");
  
  		// t, u, 1/log(d(s)) where (s, t) and (s, u) are edges in graph
  		DataSet<Tuple3<K, K, FloatValue>> twoPaths = groups
  			.groupBy(0, 1)
  			.sortGroup(2, Order.ASCENDING)
  			.reduceGroup(new GenerateGroupPairs<>())
  				.name("Generate group pairs");
  
  		// t, u, adamic-adar score
  		GroupReduceOperator<Tuple3<K, K, FloatValue>, Result<K>> scores = twoPaths
  			.groupBy(0, 1)
  			.reduceGroup(new ComputeScores<>(minimumScore, minimumRatio))
  				.name("Compute scores");
  
  		if (minimumRatio > 0.0f) {
  			// total score, number of pairs of neighbors
  			DataSet<Tuple2<FloatValue, LongValue>> sumOfScoresAndNumberOfNeighborPairs = inverseLogDegree
  				.map(new ComputeScoreFromVertex<>())
  					.setParallelism(parallelism)
  					.name("Average score")
  				.sum(0)
  				.andSum(1);
  
  			scores
  				.withBroadcastSet(sumOfScoresAndNumberOfNeighborPairs, SUM_OF_SCORES_AND_NUMBER_OF_NEIGHBOR_PAIRS);
  		}
  
  		if (mirrorResults) {
  			return scores
  				.flatMap(new MirrorResult<>())
  					.name("Mirror results");
  		} else {
  			return scores;
  		}
  	}