@Override
	public void reduce(UserIdPairWritable key, Iterator<UserRatingPairWritable> it, OutputCollector<Text, Text> output, Reporter reporter) throws IOException {
		double correlation = 0;

		// If this is the same user and not a pair, simply use 1 as correlation. We know for a fact that our algorithm
		// implementations are correct because we use Apache commons, therefor it's not a problem to simply use the
		// magic number 1 at this place. We still include it, because we want to have a complete output at the end of
		// our M/R job.
		if (key.getX() == key.getY()) {
			correlation = 1;
		}
		// If we have a user pair, prepare their ratings and pass it along to the chosen algorithm.
		else {
			UserRatingPairWritable valuePair;
			double[] xArray = new double[Coefficient.MAX_RATINGS];
			double[] yArray = new double[Coefficient.MAX_RATINGS];
			int i = 0;
	
			// Loop all ratings of the user pair and prepare the arrays for the correlation computation.
			while (it.hasNext()) {
				valuePair = it.next();
				xArray[i] = valuePair.getX();
				yArray[i] = valuePair.getY();
				i++;
			}
	
			switch (this.algorithm) {
			case p:
			case pearson:
				correlation = new PearsonsCorrelation().correlation(xArray, yArray);
				break;
			case s:
			case spearman:
				correlation = new SpearmansCorrelation().correlation(xArray, yArray);
				break;
			}
		}

		output.collect(new Text(key.toString()), new Text(String.valueOf(correlation)));
	}