@Override
		public void map(IntWritable key, Text value,
				OutputCollector<IntWritable, Text> output, Reporter reporter)
				throws IOException {
			if (key.get() < 1 || value.toString().trim() == "") {
				return;
			}

			String[] rows = value.toString().split("\\n");
			if (rows.length != key.get()) {
				throw new IllegalArgumentException(
						"Mismatched number of data points.");
			}

			RecordSet records = new RecordSet(DataPrep.NUM_FEATURES);
			for (String row : rows) {
				WeightedRecord record = new WeightedRecord(row, false,
						DataPrep.NUM_FEATURES, 1.0);
				records.append(record);
				if (record.getRelevance() == 1) {
					reporter.incrCounter(Counters.POSITIVE_DATA_POINTS, 1);
				} else if (record.getRelevance() == -1) {
					reporter.incrCounter(Counters.NEGATIVE_DATA_POINTS, 1);
				}
				reporter.incrCounter(Counters.INPUT_DATA_POINTS, 1);
				numRecords++;
			}

			AdaBoostClassifier c = AdaBoostClassifier.build(records,
					NUM_WEAK_CLASSIFIERS, 10, false);
			if (c.getWeakClassifiers().size() != NUM_WEAK_CLASSIFIERS) {
				throw new IllegalArgumentException(
						"Mismatched number of weak classifiers.");
			}
			StringBuilder sortedWeakClassifiersInText = new StringBuilder();
			for (Pair<Double, StumpClassifier> p : c
					.getSortedWeakClassifiersByWeight()) {
				sortedWeakClassifiersInText.append(p.first);
				sortedWeakClassifiersInText.append(',');
				sortedWeakClassifiersInText.append(p.second
						.getSplitFeatureIndex());
				sortedWeakClassifiersInText.append(',');
				sortedWeakClassifiersInText.append(p.second.getThreshold());
				sortedWeakClassifiersInText.append(',');
				sortedWeakClassifiersInText.append(p.second.getSplitType());
				sortedWeakClassifiersInText.append('\n');
			}
			sortedWeakClassifiersInText.setLength(sortedWeakClassifiersInText
					.length() - 1);
			// Weak classifiers have been sorted here, so no need to sort again
			// in reducer.
			output.collect(CONSTANT_ONE,
					new Text(sortedWeakClassifiersInText.toString()));

			if ((++numRecords % 100) == 0) {
				reporter.setStatus("Finished processing " + numRecords
						+ " records.");
			}
		}