SequenceFilesFromMailArchives extends AbstractJob { public static void main(String[] args) throws Exception { ToolRunner.run(new Configuration(), new SequenceFilesFromMailArchives(), args); }  void createSequenceFiles(MailOptions options); static void main(String[] args); @Override int run(String[] args); static final String[] CHUNK_SIZE_OPTION; static final String[] KEY_PREFIX_OPTION; static final String[] CHARSET_OPTION; static final String[] SUBJECT_OPTION; static final String[] TO_OPTION; static final String[] FROM_OPTION; static final String[] REFERENCES_OPTION; static final String[] BODY_OPTION; static final String[] STRIP_QUOTED_OPTION; static final String[] QUOTED_REGEX_OPTION; static final String[] SEPARATOR_OPTION; static final String[] BODY_SEPARATOR_OPTION; static final String BASE_INPUT_PATH; }
@Test public void testSequential() throws Exception { File outputDir = this.getTestTempDir("mail-archives-out"); String[] args = { "--input", inputDir.getAbsolutePath(), "--output", outputDir.getAbsolutePath(), "--charset", "UTF-8", "--keyPrefix", "TEST", "--method", "sequential", "--body", "--subject", "--separator", "" }; SequenceFilesFromMailArchives.main(args); File expectedChunkFile = new File(outputDir, "chunk-0"); String expectedChunkPath = expectedChunkFile.getAbsolutePath(); Assert.assertTrue("Expected chunk file " + expectedChunkPath + " not found!", expectedChunkFile.isFile()); Configuration conf = getConfiguration(); SequenceFileIterator<Text, Text> iterator = new SequenceFileIterator<Text, Text>(new Path(expectedChunkPath), true, conf); Assert.assertTrue("First key/value pair not found!", iterator.hasNext()); Pair<Text, Text> record = iterator.next(); File parentFile = new File(new File(new File("TEST"), "subdir"), "mail-messages.gz"); Assert.assertEquals(new File(parentFile, testVars[0][0]).toString(), record.getFirst().toString()); Assert.assertEquals(testVars[0][1] + testVars[0][2], record.getSecond().toString()); Assert.assertTrue("Second key/value pair not found!", iterator.hasNext()); record = iterator.next(); Assert.assertEquals(new File(parentFile, testVars[1][0]).toString(), record.getFirst().toString()); Assert.assertEquals(testVars[1][1] + testVars[1][2], record.getSecond().toString()); record = iterator.next(); File parentFileSubSubDir = new File(new File(new File(new File("TEST"), "subdir"), "subsubdir"), "mail-messages-2.gz"); Assert.assertEquals(new File(parentFileSubSubDir, testVars[0][0]).toString(), record.getFirst().toString()); Assert.assertEquals(testVars[0][1] + testVars[0][2], record.getSecond().toString()); Assert.assertTrue("Second key/value pair not found!", iterator.hasNext()); record = iterator.next(); Assert.assertEquals(new File(parentFileSubSubDir, testVars[1][0]).toString(), record.getFirst().toString()); Assert.assertEquals(testVars[1][1] + testVars[1][2], record.getSecond().toString()); Assert.assertFalse("Only two key/value pairs expected!", iterator.hasNext()); }
@Test public void testMapReduce() throws Exception { Path tmpDir = getTestTempDirPath(); Path mrOutputDir = new Path(tmpDir, "mail-archives-out-mr"); Configuration configuration = getConfiguration(); FileSystem fs = FileSystem.get(configuration); File expectedInputFile = new File(inputDir.toString()); String[] args = { "-Dhadoop.tmp.dir=" + configuration.get("hadoop.tmp.dir"), "--input", expectedInputFile.getAbsolutePath(), "--output", mrOutputDir.toString(), "--charset", "UTF-8", "--keyPrefix", "TEST", "--method", "mapreduce", "--body", "--subject", "--separator", "" }; SequenceFilesFromMailArchives.main(args); FileStatus[] fileStatuses = fs.listStatus(mrOutputDir.suffix("/part-m-00000")); assertEquals(1, fileStatuses.length); assertEquals("part-m-00000", fileStatuses[0].getPath().getName()); SequenceFileIterator<Text, Text> iterator = new SequenceFileIterator<Text, Text>(mrOutputDir.suffix("/part-m-00000"), true, configuration); Assert.assertTrue("First key/value pair not found!", iterator.hasNext()); Pair<Text, Text> record = iterator.next(); File parentFileSubSubDir = new File(new File(new File(new File("TEST"), "subdir"), "subsubdir"), "mail-messages-2.gz"); String expected = record.getFirst().toString(); if (SystemUtils.IS_OS_WINDOWS) { expected = expected.replace("/", "\\"); } Assert.assertEquals(new File(parentFileSubSubDir, testVars[0][0]).toString(), expected); Assert.assertEquals(testVars[0][1] + testVars[0][2], record.getSecond().toString()); Assert.assertTrue("Second key/value pair not found!", iterator.hasNext()); record = iterator.next(); expected = record.getFirst().toString(); if (SystemUtils.IS_OS_WINDOWS) { expected = expected.replace("/", "\\"); } Assert.assertEquals(new File(parentFileSubSubDir, testVars[1][0]).toString(), expected); Assert.assertEquals(testVars[1][1] + testVars[1][2], record.getSecond().toString()); File parentFile = new File(new File(new File("TEST"), "subdir"), "mail-messages.gz"); record = iterator.next(); expected = record.getFirst().toString(); if (SystemUtils.IS_OS_WINDOWS) { expected = expected.replace("/", "\\"); } Assert.assertEquals(new File(parentFile, testVars[0][0]).toString(), expected); Assert.assertEquals(testVars[0][1] + testVars[0][2], record.getSecond().toString()); Assert.assertTrue("Second key/value pair not found!", iterator.hasNext()); record = iterator.next(); expected = record.getFirst().toString(); if (SystemUtils.IS_OS_WINDOWS) { expected = expected.replace("/", "\\"); } Assert.assertEquals(new File(parentFile, testVars[1][0]).toString(), expected); Assert.assertEquals(testVars[1][1] + testVars[1][2], record.getSecond().toString()); Assert.assertFalse("Only four key/value pairs expected!", iterator.hasNext()); }
@Test public void testSequential() throws Exception { File outputDir = this.getTestTempDir("mail-archives-out"); String[] args = { "--input", inputDir.getAbsolutePath(), "--output", outputDir.getAbsolutePath(), "--charset", "UTF-8", "--keyPrefix", "TEST", "--method", "sequential", "--body", "--subject", "--separator", "" }; SequenceFilesFromMailArchives.main(args); File expectedChunkFile = new File(outputDir, "chunk-0"); String expectedChunkPath = expectedChunkFile.getAbsolutePath(); Assert.assertTrue("Expected chunk file " + expectedChunkPath + " not found!", expectedChunkFile.isFile()); Configuration conf = getConfiguration(); SequenceFileIterator<Text, Text> iterator = new SequenceFileIterator<>(new Path(expectedChunkPath), true, conf); Assert.assertTrue("First key/value pair not found!", iterator.hasNext()); Pair<Text, Text> record = iterator.next(); File parentFile = new File(new File(new File("TEST"), "subdir"), "mail-messages.gz"); Assert.assertEquals(new File(parentFile, testVars[0][0]).toString(), record.getFirst().toString()); Assert.assertEquals(testVars[0][1] + testVars[0][2], record.getSecond().toString()); Assert.assertTrue("Second key/value pair not found!", iterator.hasNext()); record = iterator.next(); Assert.assertEquals(new File(parentFile, testVars[1][0]).toString(), record.getFirst().toString()); Assert.assertEquals(testVars[1][1] + testVars[1][2], record.getSecond().toString()); record = iterator.next(); File parentFileSubSubDir = new File(new File(new File(new File("TEST"), "subdir"), "subsubdir"), "mail-messages-2.gz"); Assert.assertEquals(new File(parentFileSubSubDir, testVars[0][0]).toString(), record.getFirst().toString()); Assert.assertEquals(testVars[0][1] + testVars[0][2], record.getSecond().toString()); Assert.assertTrue("Second key/value pair not found!", iterator.hasNext()); record = iterator.next(); Assert.assertEquals(new File(parentFileSubSubDir, testVars[1][0]).toString(), record.getFirst().toString()); Assert.assertEquals(testVars[1][1] + testVars[1][2], record.getSecond().toString()); Assert.assertFalse("Only two key/value pairs expected!", iterator.hasNext()); }
@Test public void testMapReduce() throws Exception { Path tmpDir = getTestTempDirPath(); Path mrOutputDir = new Path(tmpDir, "mail-archives-out-mr"); Configuration configuration = getConfiguration(); FileSystem fs = FileSystem.get(configuration); File expectedInputFile = new File(inputDir.toString()); String[] args = { "-Dhadoop.tmp.dir=" + configuration.get("hadoop.tmp.dir"), "--input", expectedInputFile.getAbsolutePath(), "--output", mrOutputDir.toString(), "--charset", "UTF-8", "--keyPrefix", "TEST", "--method", "mapreduce", "--body", "--subject", "--separator", "" }; SequenceFilesFromMailArchives.main(args); FileStatus[] fileStatuses = fs.listStatus(mrOutputDir.suffix("/part-m-00000")); assertEquals(1, fileStatuses.length); assertEquals("part-m-00000", fileStatuses[0].getPath().getName()); SequenceFileIterator<Text, Text> iterator = new SequenceFileIterator<>(mrOutputDir.suffix("/part-m-00000"), true, configuration); Assert.assertTrue("First key/value pair not found!", iterator.hasNext()); Pair<Text, Text> record = iterator.next(); File parentFileSubSubDir = new File(new File(new File(new File("TEST"), "subdir"), "subsubdir"), "mail-messages-2.gz"); String expected = record.getFirst().toString(); if (SystemUtils.IS_OS_WINDOWS) { expected = expected.replace("/", "\\"); } Assert.assertEquals(new File(parentFileSubSubDir, testVars[0][0]).toString(), expected); Assert.assertEquals(testVars[0][1] + testVars[0][2], record.getSecond().toString()); Assert.assertTrue("Second key/value pair not found!", iterator.hasNext()); record = iterator.next(); expected = record.getFirst().toString(); if (SystemUtils.IS_OS_WINDOWS) { expected = expected.replace("/", "\\"); } Assert.assertEquals(new File(parentFileSubSubDir, testVars[1][0]).toString(), expected); Assert.assertEquals(testVars[1][1] + testVars[1][2], record.getSecond().toString()); File parentFile = new File(new File(new File("TEST"), "subdir"), "mail-messages.gz"); record = iterator.next(); expected = record.getFirst().toString(); if (SystemUtils.IS_OS_WINDOWS) { expected = expected.replace("/", "\\"); } Assert.assertEquals(new File(parentFile, testVars[0][0]).toString(), expected); Assert.assertEquals(testVars[0][1] + testVars[0][2], record.getSecond().toString()); Assert.assertTrue("Second key/value pair not found!", iterator.hasNext()); record = iterator.next(); expected = record.getFirst().toString(); if (SystemUtils.IS_OS_WINDOWS) { expected = expected.replace("/", "\\"); } Assert.assertEquals(new File(parentFile, testVars[1][0]).toString(), expected); Assert.assertEquals(testVars[1][1] + testVars[1][2], record.getSecond().toString()); Assert.assertFalse("Only four key/value pairs expected!", iterator.hasNext()); }