CascadingTupleInputFormat extends FileInputFormat<ITuple, NullWritable> implements Serializable { public static void setSerializations(Configuration conf) { String serializations = conf.get(HADOOP_IO_SERIALIZATIONS); LinkedList<String> list = new LinkedList<String>(); if(serializations != null && !serializations.isEmpty()) Collections.addAll(list, serializations.split(",")); String writable = WritableSerialization.class.getName(); String tuple = TupleSerialization.class.getName(); list.remove(writable); list.remove(tuple); list.addFirst(writable); list.addFirst(tuple); conf.set(HADOOP_IO_SERIALIZATIONS, Util.join(list, ",")); } CascadingTupleInputFormat(String schemaName, String... fieldNames); static void setSerializations(Configuration conf); @Override RecordReader<ITuple, NullWritable> createRecordReader(InputSplit split, TaskAttemptContext ctx);  }
@Test public void test() throws Exception { MapOnlyJobBuilder builder = new MapOnlyJobBuilder(getConf()); CascadingTupleInputFormat.setSerializations(getConf()); InputFormat<ITuple, NullWritable> iF = new CascadingTupleInputFormat("logs", "day", "month", "year", "count", "metric", "value"); builder.addInput(new Path("src/test/resources/cascading-binary"), iF, new MapOnlyMapper<ITuple, NullWritable, Text, NullWritable>() { @Override protected void map(ITuple key, NullWritable value, Context context) throws IOException, InterruptedException { context.write(new Text(key.toString()), NullWritable.get()); } }); builder.setOutput(new Path(OUTPUT), new HadoopOutputFormat(TextOutputFormat.class), Text.class, NullWritable.class); Job job = builder.createJob(); try { assertRun(job); } finally { builder.cleanUpInstanceFiles(); } String expectedOutput = "{\"day\":20,\"month\":10,\"year\":2012,\"count\":97,\"metric\":\"ALL\",\"value\":\"\"}\n" + "{\"day\":21,\"month\":10,\"year\":2012,\"count\":717,\"metric\":\"ALL\",\"value\":\"\"}\n" + "{\"day\":22,\"month\":10,\"year\":2012,\"count\":186,\"metric\":\"ALL\",\"value\":\"\"}"; assertEquals(expectedOutput, Files.toString(new File(OUTPUT, "part-m-00000"), Charset.defaultCharset()).trim()); HadoopUtils.deleteIfExists(FileSystem.get(getConf()), new Path(OUTPUT)); }