IndexerJob { public void startIndexer(String path, String finalDestination, int numOfShards) throws IOException { JobConf jobConf = new JobConf(IndexerJob.class); jobConf.setJobName("indexer"); jobConf.setMapRunnerClass(Indexer.class); jobConf.setInputFormat(SequenceFileInputFormat.class); Path input = new Path(path); FileInputFormat.setInputPaths(jobConf, input); FileOutputFormat.setOutputPath(jobConf, new Path(finalDestination)); jobConf.set("finalDestination", finalDestination); jobConf.setSpeculativeExecution(false); FileSystem fs = FileSystem.get(input.toUri(), jobConf); FileStatus[] status = fs.globStatus(input); long size = 0; for (FileStatus fileStatus : status) { size += fileStatus.getLen(); } long optimalSplisize = size / numOfShards; jobConf.set("mapred.min.split.size", "" + optimalSplisize); jobConf.set("mapred.child.java.opts", "-Xmx2G"); jobConf.setNumMapTasks(1); jobConf.setNumReduceTasks(0); JobClient.runJob(jobConf); }  static void main(String[] args); void startIndexer(String path, String finalDestination, int numOfShards);  }
@Test public void testStartJob() throws Exception { IndexerJob indexerJob = new IndexerJob(); File tmp = new File("./build/extras/indexing/tmp/IndexerJobTest"); tmp.mkdirs(); File in = new File(tmp, "in"); File out = new File(tmp, "out"); FileUtil.fullyDelete(out); String alicePath = SearchPathUtil.findPath("sample-data/texts/alice.txt", "../../sample-data/texts/alice.txt"); String sampleText = SequenceFileCreator.getSampleText(alicePath); SequenceFileCreator creator = new SequenceFileCreator(); creator.create(in.getAbsolutePath(), sampleText, 100000); indexerJob.startIndexer(in.getAbsolutePath(), out.getAbsolutePath(), 5); }