TokenizerMapper extends Mapper<Object, Text, Text, TextArrayWritable> { List<ImmutablePair<String, List<String>>> parse(String value) { List<ImmutablePair<String, List<String>>> result = newArrayList(); StringTokenizer tokenizer = new StringTokenizer(value, "\n"); while (tokenizer.hasMoreElements()) { String line = (String) tokenizer.nextElement(); String[] parsed = line.split(",", -1); String origin = parsed[0]; String dest = parsed[1]; String flightDate = parsed[2]; String flightNum = parsed[3]; String depTime = parsed[4]; String arrDelay = parsed[5]; String cancelled = parsed[6]; List<String> strings = newArrayList(flightNum, depTime, arrDelay); if (isNotCancelled(cancelled) && isNotBlank(arrDelay)) { result.add(of(merge(origin, dest, flightDate, DepartureType.build(Integer.valueOf(depTime)).name()), strings)); } } return result; }   static final String SEPARATOR; }
@Test public void givenString_whenCodes_whenTokenized() { TokenizerMapper tokenizerMapper = new TokenizerMapper(); List<ImmutablePair<String, List<String>>> pairs = tokenizerMapper.parse("LAX,SFO,2008-01-03,593,2321,203.00,0.00\n" + "LAX,SFO,2008-01-03,618,,,1.00\n"); assertTrue(pairs.get(0).left.equals("LAX#SFO#2008-01-03#AFTER_12")); }