Class MapRedScan

src/com/winvector/logistic/mr/MapRedScan.java:27–236

import com.winvector.util.SerialUtils;

public final class MapRedScan {
	public static final String BURSTERSERFIELD = "MapRedScan.BursterScan";
	private static final String FORMULAFIELD = "MapRedScan.Formula";
	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
		public void reduce(final LongWritable key, final Iterable<WritableVariableList> values, final Context context) throws IOException, InterruptedException {
			WritableVariableList r = null;
			for(final WritableVariableList vi: values) {
				if(r==null) {
					r = WritableVariableList.copy(vi);
				} else {
					r.mergeVariableDefs(vi);