throw new DataUtilException("The comparator parameter can not be a null value");
}
// reduce all data
FileInputStream inputStream;
OutputCollector collector = new OutputCollector(outputFile);
QueueReader queueReader;
// Aggregate all items matching the comparator and call the Reducable callback
try {
inputStream = new FileInputStream(inputFile);
BufferedReader br = new BufferedReader(new InputStreamReader(inputStream,
DataUtilDefaults.charSet));
queueReader = new QueueReader(br);
// reduce data
String prev = null;
String curr;
ArrayList<String> items = new ArrayList<String>();
while(true) {
curr = queueReader.readLine();
if(curr == null) {
// no more data
reduceItems(collector, reducer, items);
break;
} else if(prev == null) {
// first row in a new batch
items.add(curr);
prev = curr;
} else if(comparator.compare(prev, curr) == 0) {
// same keys
items.add(curr);
prev = curr;
} else {
// different keys
queueReader.push(curr);
reduceItems(collector, reducer, items);
items.clear();
prev = null;
}
}
collector.close();
inputStream.close();
} catch (IOException e) {
throw new DataUtilException(e);
} catch (MapReduceException e) {
throw new DataUtilException("Irrecoverable reduce operation", e);