TeraSort extends Configured implements Tool { public int run(String[] args) throws Exception { if (args.length != 2) { usage(); return 2; } LOG.info("starting"); Job job = Job.getInstance(getConf()); Path inputDir = new Path(args[0]); Path outputDir = new Path(args[1]); boolean useSimplePartitioner = getUseSimplePartitioner(job); TeraInputFormat.setInputPaths(job, inputDir); FileOutputFormat.setOutputPath(job, outputDir); job.setJobName("TeraSort"); job.setJarByClass(TeraSort.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(Text.class); job.setInputFormatClass(TeraInputFormat.class); job.setOutputFormatClass(TeraOutputFormat.class); if (useSimplePartitioner) { job.setPartitionerClass(SimplePartitioner.class); } else { long start = System.currentTimeMillis(); Path partitionFile = new Path(outputDir, TeraInputFormat.PARTITION_FILENAME); URI partitionUri = new URI(partitionFile.toString() + "#" + TeraInputFormat.PARTITION_FILENAME); try { TeraInputFormat.writePartitionFile(job, partitionFile); } catch (Throwable e) { LOG.error("{}", e.getMessage(), e); return -1; } job.addCacheFile(partitionUri); long end = System.currentTimeMillis(); System.out.println("Spent " + (end - start) + "ms computing partitions."); job.setPartitionerClass(TotalOrderPartitioner.class); } job.getConfiguration().setInt("dfs.replication", getOutputReplication(job)); int ret = job.waitForCompletion(true) ? 0 : 1; LOG.info("done"); return ret; }  static boolean getUseSimplePartitioner(JobContext job); static void setUseSimplePartitioner(Job job, boolean value); static int getOutputReplication(JobContext job); static void setOutputReplication(Job job, int value); int run(String[] args); static void main(String[] args);  }
@Test public void testTeraSortWithLessThanTwoArgs() throws Exception { String[] args = new String[1]; assertThat(new TeraSort().run(args)).isEqualTo(2); }