public void reduce(Text key, Iterator<Text> values, OutputCollector<Text, Text> output, Reporter reporter) throws IOException {
			// allocate and initialize 2D array of category by district
			int[][] crimes = new int[categories.size()][districts.size()];
			for (int i = 0; i < categories.size(); i++) {
				for (int j = 0; j < districts.size(); j++) {
					crimes[i][j] = 0;
				}
			}
			// create the heat map of crimes in the category/district array
			while (values.hasNext()) {
				String crime = values.next().toString();
				String[] cols = DataFile.getColumns(crime);
				if (cols.length == 2) {
					if (categoryLookup.containsKey(cols[1])) {
						if (districtLookup.containsKey(cols[0])) {
							int cat = categoryLookup.get(cols[1]);
							int dist = districtLookup.get(cols[0]);
							crimes[cat][dist]++;
						} else {
							log.warning(MessageFormat.format("District {0} not found.", new Object[]{cols[0]}));
						}
					} else {
						log.warning(MessageFormat.format("Category {0} not found.", new Object[]{cols[1]}));
					}
				} else {
					log.warning(MessageFormat.format("Input {0} was in unexpected format", new Object[]{crime}));
				}
			}
			// serialize the non zero entries as a triplet of category index, district index, and total crimes per day
			for (int i = 0; i < categories.size(); i++) {
				for (int j = 0; j < districts.size(); j++) {
					if (crimes[i][j] > 0) {
						StringBuffer sv = new StringBuffer();
						sv.append(new Integer(i).toString());
						sv.append(",");
						sv.append(new Integer(j).toString());
						sv.append(",");
						sv.append(new Integer(crimes[i][j]));
						Text tv = new Text();
						tv.set(sv.toString());
						output.collect(key, tv);
					}
				}
			}
		}