FormatProfiler extends Configured implements Tool { public void createJobConf(JobConf conf, String[] args) throws IOException { log.info("Loading paths..."); String line = null; List<Path> paths = new ArrayList<Path>(); BufferedReader br = new BufferedReader( new FileReader( args[ 0 ] ) ); while( ( line = br.readLine() ) != null ) { paths.add( new Path( line ) ); } br.close(); log.info("Setting paths..."); FileInputFormat.setInputPaths( conf, paths.toArray(new Path[] {}) ); log.info("Set "+paths.size()+" InputPaths"); FileOutputFormat.setOutputPath( conf, new Path( args[ 1 ] ) ); conf.setJobName( "NaniteFormatProfiler-"+new File(args[ 0 ]).getName() + "_" + System.currentTimeMillis() ); conf.setInputFormat( ArchiveFileInputFormat.class ); conf.setMapperClass( FormatProfilerMapper.class ); conf.setReducerClass( FormatProfilerReducer.class ); conf.setOutputFormat( TextOutputFormat.class ); conf.set( "map.output.key.field.separator", "" ); conf.setOutputKeyClass( Text.class ); conf.setOutputValueClass( Text.class ); conf.setMapOutputValueClass( Text.class ); conf.setUserClassesTakesPrecedence(true); int timeoutMins = 30; conf.setInt("mapred.task.timeout", timeoutMins*60*1000); conf.set("mapred.user.jobconf.limit", "104857600"); Config config = ConfigFactory.load(); conf.setNumReduceTasks(config.getInt("warc.hadoop.num_reducers")); }  void createJobConf(JobConf conf, String[] args); int run( String[] args ); static void main( String[] args );  }
@Test public void testFullFormatProfilerJob() throws Exception { log.info("Checking input file is present..."); Path[] inputFiles = FileUtil.stat2Paths(getFileSystem().listStatus( input, new OutputLogFilter())); Assert.assertEquals(1, inputFiles.length); String[] args = {"src/test/resources/test-inputs.txt", this.output.getName()}; @SuppressWarnings("unused") Config config = ConfigFactory.load(); FormatProfiler wir = new FormatProfiler(); log.info("Setting up job config..."); JobConf conf = this.mrCluster.createJobConf(); wir.createJobConf(conf, args); conf.set( "mapred.reduce.tasks.speculative.execution", "false" ); log.info("Running job..."); JobClient.runJob(conf); log.info("Job finished, checking the results..."); Path[] outputFiles = FileUtil.stat2Paths( getFileSystem().listStatus( output ) ); for( Path output : outputFiles ) { log.info(" --- output : "+output); if( getFileSystem().isFile(output) ) { InputStream is = getFileSystem().open(output); BufferedReader reader = new BufferedReader(new InputStreamReader(is)); String line = null; while( ( line = reader.readLine()) != null ) { log.info(line); } reader.close(); } else { log.info(" --- ...skipping directory..."); } } }