TAReducer extends Reducer<TAKey, L3TemporalBinWithIndex, Text, TAPoint> implements Configurable { @Override protected void reduce(TAKey taKey, Iterable<L3TemporalBinWithIndex> bins, Context context) throws IOException, InterruptedException { final DateFormat dateFormat = DateUtils.createDateFormat(DATE_PATTERN); LOGGER.info("handling bins for " + taKey + " ..."); final TAConfig.RegionConfiguration region = regions[taKey.getRegionId()]; FileSystem fs = new Path(outputDirPath).getFileSystem(conf); int numFeatures = outputFeatureNames.size(); PrintWriter[] writers = new PrintWriter[numFeatures + 1]; StringBuffer[] lines = new StringBuffer[numFeatures + 1]; ArrayList<float[]> lineValuesList = new ArrayList<>(); Map<Long,Integer> columnOfBinIndexMap = new HashMap<>(); L3TemporalBin outputBin = null; long currentTime = -1; for (L3TemporalBinWithIndex bin : bins) { if (currentTime == -1) { currentTime = bin.getTime(); LOGGER.info("Handling first time step " + dateFormat.format(new Date(currentTime))); outputBin = (L3TemporalBin) binManager.createTemporalBin(-1); initialiseCsvWriting(fs, region, numFeatures, writers, lines); } else if (bin.getTime() != currentTime) { writeAggregatedRecord(context, region, dateFormat, currentTime, outputBin); outputBin = (L3TemporalBin) binManager.createTemporalBin(-1); final String timeString = dateFormat.format(new Date(currentTime)); writeCurrentLines(timeString, numFeatures, writers, lineValuesList); currentTime = bin.getTime(); LOGGER.info("Handling time step " + dateFormat.format(new Date(currentTime))); } final float[] values = getValueContainerForBinIndex(columnOfBinIndexMap, lineValuesList, numFeatures, bin.getIndex()); for (int i = 0; i < numFeatures; ++i) { values[i] = bin.getFeatureValues()[i]; } values[numFeatures] = bin.getNumObs(); binManager.aggregateTemporalBin(bin, outputBin); } writeCurrentLines(dateFormat.format(currentTime), numFeatures, writers, lineValuesList); if (currentTime != -1) { writeAggregatedRecord(context, region, dateFormat, currentTime, outputBin); for (int i = 0; i < numFeatures + 1; ++i) { writers[i].flush(); writers[i].close(); } } }  @Override void setConf(Configuration conf); @Override Configuration getConf(); static final String DATE_PATTERN; static final Logger LOGGER; }
@Test public void testReduce() throws Exception { List<L3TemporalBin> temporalBins = new ArrayList<L3TemporalBin>(); temporalBins.add(createTBin(1, 10, 0.2f, 0.15f, 0.3f)); temporalBins.add(createTBin(2, 7, 0.4f, 0.25f, 0.3f)); temporalBins.add(createTBin(4, 6, 0.6f, 0.35f, 0.3f)); L3TemporalBin outputBin = (L3TemporalBin) taReducer.binManager.createTemporalBin(-1); for (L3TemporalBin bin : temporalBins) { taReducer.binManager.aggregateTemporalBin(bin, outputBin); } TAPoint taPoint = new TAPoint("northsea", taReducer.minDate, taReducer.maxDate, outputBin); assertNotNull(taPoint); assertEquals("northsea", taPoint.getRegionName()); assertEquals("2010-01-01", taPoint.getStartDate()); assertEquals("2010-01-10", taPoint.getStopDate()); TemporalBin temporalBin = taPoint.getTemporalBin(); assertNotNull(temporalBin); assertEquals(1 + 2 + 4, temporalBin.getNumPasses()); assertEquals(10 + 7 + 6, temporalBin.getNumObs()); float[] featureValues = temporalBin.getFeatureValues(); assertEquals(3, featureValues.length); assertEquals(8.4f, featureValues[0], 1e-5f); assertEquals(5.35f, featureValues[1], 1e-5f); assertEquals(23f, featureValues[2], 1e-5f); assertEquals("TAPoint{regionName=northsea, startDate=2010-01-01, stopDate=2010-01-10, temporalBin=L3TemporalBin{index=-1, numObs=23, numPasses=7, featureValues=[8.400001, 5.35, 23.0]}}", taPoint.toString()); }