SequenceFileConverterMapper extends
        Mapper<Writable, Writable, Text, BehemothDocument> { @Override protected void map(Writable key, Writable value, Context context) throws IOException, InterruptedException { BehemothDocument doc = new BehemothDocument(); doc.setUrl(key.toString()); DataOutputBuffer out = new DataOutputBuffer(); value.write(out); doc.setContent(out.getData()); context.write(new Text(key.toString()), doc); }    }
@Test public void test() throws Exception { Configuration conf = new Configuration(); FileSystem fs = FileSystem.get(conf); File tmpDir = new File(System.getProperty("java.io.tmpdir")); File tmpLoc = new File(tmpDir, "sfcmt"); tmpLoc.mkdirs(); Path in = new Path(tmpLoc.getAbsolutePath(), "foo"); SequenceFile.Writer writer = SequenceFile.createWriter(fs, conf, in, Text.class, Text.class); for (int i = 0; i < 100; i++) { writer.append(new Text(String.valueOf(i)), new Text( "This is document " + i)); } writer.close(); SequenceFileConverterMapper mapper = new SequenceFileConverterMapper(); SequenceFileIterator<Text, Text> iter = new SequenceFileIterator<Text, Text>( in, true, conf); DummyRecordWriter out = new DummyRecordWriter(); Mapper.Context context = mapper.new Context(conf, new TaskAttemptID(), null, out, null, new DummyStatusReporter(), null); while (iter.hasNext()) { Pair<Text, Text> next = iter.next(); mapper.map(next.getFirst(), next.getSecond(), context); } assertEquals(100, out.keys.size()); }
@Test public void testWritable() throws Exception { Configuration conf = new Configuration(); FileSystem fs = FileSystem.get(conf); File tmpDir = new File(System.getProperty("java.io.tmpdir")); File tmpLoc = new File(tmpDir, "sfcmt"); tmpLoc.mkdirs(); Path in = new Path(tmpLoc.getAbsolutePath(), "foo"); SequenceFile.Writer writer = SequenceFile.createWriter(fs, conf, in, Text.class, MyWritable.class); for (int i = 0; i < 100; i++) { writer.append(new Text(String.valueOf(i)), new MyWritable(i, "document " + i)); } writer.close(); SequenceFileConverterMapper mapper = new SequenceFileConverterMapper(); SequenceFileIterator<Text, Text> iter = new SequenceFileIterator<Text, Text>( in, true, conf); DummyRecordWriter out = new DummyRecordWriter(); Mapper.Context context = mapper.new Context(conf, new TaskAttemptID(), null, out, null, new DummyStatusReporter(), null); while (iter.hasNext()) { Pair<Text, Text> next = iter.next(); mapper.map(next.getFirst(), next.getSecond(), context); } assertEquals(100, out.keys.size()); int i = 0; for (BehemothDocument doc : out.vals) { byte[] bytes = doc.getContent(); DataInput input = new DataInputStream(new ByteArrayInputStream( bytes)); MyWritable writable = new MyWritable(); writable.readFields(input); assertEquals(writable.i, i); assertEquals(writable.in, "document " + i); i++; } }