@Test
public void testSequenceFileFromDirectoryMapReduce() throws Exception {

  Configuration conf = getConfiguration();
  FileSystem fs = FileSystem.get(conf);
  // create
  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");
}