@Override
	public void reduce(LongWritable key, Iterator<DoubleArrayWritable> it, OutputCollector<Text, Text> output, Reporter reporter) throws IOException {
		DoubleWritable[] caster = new DoubleWritable[Coefficient.MAX_RATINGS];
		double[] currentUserRatings = new double[Coefficient.MAX_RATINGS];
		double[] otherUserRatings = new double[Coefficient.MAX_RATINGS];
		double correlation = 0;
		int i = 0;

		while (it.hasNext()) {
			// The current user we want to compute.
			if (key.get() == i) {
				// Get the DoubleWritable[] and cast each value to a primitive double value.
				caster = it.next().get();
				for (int j = 0; j < caster.length; j++) {
					currentUserRatings[j] = caster[j].get();
				}

				// Correlation for ourself is always 1.
				output.collect(new Text(key.toString() + "," + key.toString()), new Text("1"));
			}
			// The users after the current users we want to compute. We've already computed users before the current
			// user in a preceding reduce task.
			else if (i > key.get()) {
				caster = it.next().get();
				for (int j = 0; j < caster.length; j++) {
					otherUserRatings[j] = caster[j].get();
				}

				switch (this.algorithm) {
				case p:
				case pearson:
					correlation = new PearsonsCorrelation().correlation(currentUserRatings, otherUserRatings);
					break;
				case s:
				case spearman:
					correlation = new SpearmansCorrelation().correlation(currentUserRatings, otherUserRatings);
					break;
				}

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

			i++;
		}
	}