@Override
	protected void reduce(IntWritable inKey,
			Iterable<IntWritable> inValues, Context context)
			throws IOException, InterruptedException {

		Configuration conf = context.getConfiguration();
		int topResults = Integer.parseInt(conf.get("inlinks.top_results"));

		for (IntWritable inValue : inValues) {
			int page = inValue.get(), pageInLinks = inKey.get();

			if (topN.size() < topResults || pageInLinks >= topN.peek().getKey()) {
				topN.add(new AbstractMap.SimpleEntry<Integer, Integer>(pageInLinks, page));
				if (topN.size() > topResults) {
					topN.poll();
				}
			}
		}
	}