StreamXmlRecordReader extends StreamBaseRecordReader { public StreamXmlRecordReader(FSDataInputStream in, FileSplit split, TaskAttemptContext context, Configuration conf, FileSystem fs) throws IOException { super(in, split, context, conf, fs); beginMark_ = checkJobGet(CONF_NS + "begin"); endMark_ = checkJobGet(CONF_NS + "end"); maxRecSize_ = conf_.getInt(CONF_NS + "maxrec", 50 * 1000); lookAhead_ = conf_.getInt(CONF_NS + "lookahead", 2 * maxRecSize_); synched_ = false; slowMatch_ = conf_.getBoolean(CONF_NS + "slowmatch", false); if (slowMatch_) { beginPat_ = makePatternCDataOrMark(beginMark_); endPat_ = makePatternCDataOrMark(endMark_); } init(); } StreamXmlRecordReader(FSDataInputStream in, FileSplit split,
      TaskAttemptContext context, Configuration conf, FileSystem fs); final void init(); synchronized boolean next(Text key, Text value); void seekNextRecordBoundary(); @Override Text getCurrentKey(); @Override Text getCurrentValue(); @Override void initialize(InputSplit arg0, TaskAttemptContext arg1); @Override boolean nextKeyValue();  }
@Test public void testStreamXmlRecordReader() throws Exception { Job job = Job.getInstance(); Configuration conf = job.getConfiguration(); job.setJarByClass(TestStreamXmlRecordReader.class); job.setMapperClass(Mapper.class); conf.set("stream.recordreader.class", "org.apache.hadoop.streaming.mapreduce.StreamXmlRecordReader"); conf.set("stream.recordreader.begin", "<PATTERN>"); conf.set("stream.recordreader.end", "</PATTERN>"); job.setInputFormatClass(StreamInputFormat.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(Text.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(Text.class); FileInputFormat.addInputPath(job, new Path("target/input.xml")); OUTPUT_DIR = new Path("target/output"); fs = FileSystem.get(conf); if (fs.exists(OUTPUT_DIR)) { fs.delete(OUTPUT_DIR, true); } FileOutputFormat.setOutputPath(job, OUTPUT_DIR); boolean ret = job.waitForCompletion(true); assertEquals(true, ret); checkOutput(); }
@Test public void testStreamXmlRecordReader() throws Exception { Job job = new Job(); Configuration conf = job.getConfiguration(); job.setJarByClass(TestStreamXmlRecordReader.class); job.setMapperClass(Mapper.class); conf.set("stream.recordreader.class", "org.apache.hadoop.streaming.mapreduce.StreamXmlRecordReader"); conf.set("stream.recordreader.begin", "<PATTERN>"); conf.set("stream.recordreader.end", "</PATTERN>"); job.setInputFormatClass(StreamInputFormat.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(Text.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(Text.class); FileInputFormat.addInputPath(job, new Path("target/input.xml")); OUTPUT_DIR = new Path("target/output"); fs = FileSystem.get(conf); if (fs.exists(OUTPUT_DIR)) { fs.delete(OUTPUT_DIR, true); } FileOutputFormat.setOutputPath(job, OUTPUT_DIR); boolean ret = job.waitForCompletion(true); assertEquals(true, ret); checkOutput(); }