SequenceFilesFromDirectory extends AbstractJob { public static void main(String[] args) throws Exception { ToolRunner.run(new SequenceFilesFromDirectory(), args); }  static void main(String[] args); @Override int run(String[] args); static final String[] FILE_FILTER_CLASS_OPTION; static final String[] KEY_PREFIX_OPTION; static final String BASE_INPUT_PATH; }
@Test public void testSequenceFileFromDirectoryBasic() throws Exception { Configuration configuration = getConfiguration(); FileSystem fs = FileSystem.get(configuration); Path tmpDir = this.getTestTempDirPath(); Path inputDir = new Path(tmpDir, "inputDir"); fs.mkdirs(inputDir); Path outputDir = new Path(tmpDir, "outputDir"); Path outputDirRecursive = new Path(tmpDir, "outputDirRecursive"); Path inputDirRecursive = new Path(tmpDir, "inputDirRecur"); fs.mkdirs(inputDirRecursive); createFilesFromArrays(configuration, inputDir, DATA1); SequenceFilesFromDirectory.main(new String[]{ "--input", inputDir.toString(), "--output", outputDir.toString(), "--chunkSize", "64", "--charset", Charsets.UTF_8.name(), "--keyPrefix", "UID", "--method", "sequential"}); checkChunkFiles(configuration, outputDir, DATA1, "UID"); createRecursiveDirFilesFromArrays(configuration, inputDirRecursive, DATA2); FileStatus fstInputPath = fs.getFileStatus(inputDirRecursive); String dirs = HadoopUtil.buildDirList(fs, fstInputPath); System.out.println("\n\n ----- recursive dirs: " + dirs); SequenceFilesFromDirectory.main(new String[]{ "--input", inputDirRecursive.toString(), "--output", outputDirRecursive.toString(), "--chunkSize", "64", "--charset", Charsets.UTF_8.name(), "--keyPrefix", "UID", "--method", "sequential"}); checkRecursiveChunkFiles(configuration, outputDirRecursive, DATA2, "UID"); }
@Test public void testSequenceFileFromDirectoryMapReduce() throws Exception { Configuration conf = getConfiguration(); FileSystem fs = FileSystem.get(conf); Path tmpDir = this.getTestTempDirPath(); Path inputDir = new Path(tmpDir, "inputDir"); fs.mkdirs(inputDir); Path inputDirRecur = new Path(tmpDir, "inputDirRecur"); fs.mkdirs(inputDirRecur); Path mrOutputDir = new Path(tmpDir, "mrOutputDir"); Path mrOutputDirRecur = new Path(tmpDir, "mrOutputDirRecur"); createFilesFromArrays(conf, inputDir, DATA1); SequenceFilesFromDirectory.main(new String[]{ "-Dhadoop.tmp.dir=" + conf.get("hadoop.tmp.dir"), "--input", inputDir.toString(), "--output", mrOutputDir.toString(), "--chunkSize", "64", "--charset", Charsets.UTF_8.name(), "--method", "mapreduce", "--keyPrefix", "UID", "--fileFilterClass", "org.apache.mahout.text.TestPathFilter" }); checkMRResultFiles(conf, mrOutputDir, DATA1, "UID"); createRecursiveDirFilesFromArrays(conf, inputDirRecur, DATA2); FileStatus fst_input_path = fs.getFileStatus(inputDirRecur); String dirs = HadoopUtil.buildDirList(fs, fst_input_path); logger.info("\n\n ---- recursive dirs: {}", dirs); SequenceFilesFromDirectory.main(new String[]{ "-Dhadoop.tmp.dir=" + conf.get("hadoop.tmp.dir"), "--input", inputDirRecur.toString(), "--output", mrOutputDirRecur.toString(), "--chunkSize", "64", "--charset", Charsets.UTF_8.name(), "--method", "mapreduce", "--keyPrefix", "UID", "--fileFilterClass", "org.apache.mahout.text.TestPathFilter" }); checkMRResultFilesRecursive(conf, mrOutputDirRecur, DATA2, "UID"); }