@Override
	public void map(ShortWritable inKey, FloatArrayWritable inValue,
			Context context) throws IOException, InterruptedException {

		Configuration conf = context.getConfiguration();
		short blockSize = Short.parseShort(conf.get("pagerank.block_size"));
		int topResults = Integer.parseInt(conf.get("pagerank.top_results"));

		Writable[] vStripe = inValue.get();
		for (int i = 0; i < vStripe.length; i++) {
			int page = 1 + (inKey.get() - 1) * blockSize + i;
			float pageRank = ((FloatWritable) vStripe[i]).get();

			// The elements in the queue are sorted (in non-decreasing order) by
			// PageRank. The queue is filled up until it contains topResults
			// elements. Then, a new element will be added only if its PageRank
			// is greater than the lowest PageRank in the queue. If the queue is
			// full and a new element is added, the one with the lowest PageRank
			// is removed from the queue.
			if (topN.size() < topResults || pageRank >= topN.peek().getKey()) {
				topN.add(new AbstractMap.SimpleEntry<Float, Integer>(pageRank, page));
				if (topN.size() > topResults) {
					topN.poll();
				}
			}
		}
	}