public void sort() throws IOException {
		
		Partitioner<Text> partitioner = new Partitioner<Text>();
		
		BufferedReader[] inputArray = new BufferedReader[this.fileCounter];
		BufferedOutputStream[] outputArray = new BufferedOutputStream[this.partitionNum];
		
		/* Initialization */
		for (int i = 0; i < this.fileCounter; i++) {
			String filename = String.format("%s/%s-%s-tmp-%d.tmp",
					task.getFilePrefix(), task.getJobId(), task.getTaskId(), i);
			inputArray[i] = new BufferedReader(new FileReader(filename));
		}
		
		for (int i = 0; i < this.partitionNum; i++) {
			String filename = String.format("%s/%s-%s-%d",
					task.getFilePrefix(), task.getJobId(), task.getTaskId(), i);
			outputArray[i] = new BufferedOutputStream(new FileOutputStream(filename));
		}
		
		Queue<FileKeyValue> heap = new PriorityQueue<FileKeyValue>();
		
		/* External sort */
		for (int i = 0; i < this.fileCounter; i++) {
			String line = inputArray[i].readLine();
			String[] comp = line.split("\t");
			heap.offer(new FileKeyValue(i, new Text(comp[0]), new IntWritable(comp[1])));
		}
		
		while (!heap.isEmpty()) {
			
			FileKeyValue towrite = heap.poll();
			
			int parNum = partitioner.getPartition(towrite.getPair().getKey(), this.partitionNum);
			
			String buff = String.format("%s\t%s\n", towrite.getPair().getKey().toString(), towrite.getPair().getValue().toString());
			
			outputArray[parNum].write(buff.getBytes());
			
			String line = inputArray[towrite.getFileSEQ()].readLine();
			if (line != null) {
				String[] comp = line.split("\t");
				heap.offer(new FileKeyValue(towrite.getFileSEQ(), new Text(comp[0]), new IntWritable(comp[1])));
			}
		}
		
		/* Close */
		for (int i = 0; i < this.fileCounter; i++) {
			inputArray[i].close();;
		}
		
		for (int i = 0; i < this.partitionNum; i++) {
			outputArray[i].flush();
			outputArray[i].close();
		}
		
		/* delete temp file */
		for (int i = 0; i < this.fileCounter; i++) {
			String filename = String.format("%s/%s-%s-tmp-%d.tmp",
					task.getFilePrefix(), task.getJobId(), task.getTaskId(), i);
			File tmpFile = new File(filename);
			tmpFile.delete();
		}
		
	}