Class DefMapper

src/com/winvector/logistic/mr/MapRedScan.java:32–74

	private static final String DEFFIELD = "MapRedScan.DefFieldName";
	
	public static final class DefMapper extends Mapper<LongWritable,Text,LongWritable,WritableVariableList> {
		// config
		private LineBurster burster = null;
		// derived
		private Log log = null;
		private String hostDescr = null;
		// result
		private WritableVariableList accum = null;
		
		@Override
		public void setup(final Context context) throws IOException {
			log = LogFactory.getLog(DefMapper.class);
			hostDescr = WritableUtils.hostDescr();			
			log.info(".setup() " + hostDescr); 
			// read side-channel configuration
			accum = null;
			try {
				burster = SerialUtils.readSerialiazlabeFromString(context.getConfiguration().get(MapRedScan.BURSTERSERFIELD));
				final String formulaStr = context.getConfiguration().get(FORMULAFIELD);
				final Formula f = new Formula(formulaStr);
				accum = new WritableVariableList();
				accum.readyForDefTracking(f);
			} catch (Exception e) {
				throw new IOException(e.toString());
			}
		}

		@Override
		public void map(final LongWritable key, final Text value, final Context context) {
			final BurstMap parsed = burster.parse(value.toString());
			if(!parsed.isEmpty()) {
				accum.trackVariableDefsFromRow(parsed);
			}
		}
		
		@Override
		public void cleanup(final Context context) throws IOException, InterruptedException {
			context.write(new LongWritable(0),accum);
			log.info(".cleanup() " + hostDescr);
			burster = null;
			accum = null;
		}
	}
	
	public static final class DefReducer extends Reducer<LongWritable,WritableVariableList,LongWritable,WritableVariableList> {
		@Override