@Test public void map_reduce() throws Exception { Job job = newJob(); job.setInputFormatClass(TextInputFormat.class); job.setOutputFormatClass(TextOutputFormat.class); job.setMapperClass(WordCountMapper.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(LongWritable.class); job.setSortComparatorClass(Text.Comparator.class); job.setGroupingComparatorClass(Text.Comparator.class); job.setReducerClass(WordCountReducer.class); job.setNumReduceTasks(1); job.setOutputKeyClass(Text.class); job.setOutputValueClass(LongWritable.class); File inputDir = folder.newFolder(); File inputFile = new File(inputDir, "input.txt"); write(inputFile, new String[] { "a b c d", "a a b c", "c", }); File outputDir = folder.newFolder(); outputDir.delete(); FileInputFormat.setInputPaths(job, new Path(inputFile.toURI())); FileOutputFormat.setOutputPath(job, new Path(outputDir.toURI())); assertThat(new SimpleJobRunner().run(job), is(true)); assertThat(toMap(read(outputDir)), is(map(new String[] { "a", "3", "b", "2", "c", "3", "d", "1", }))); }
@Override public boolean run(Job job) throws InterruptedException { job.setJobID(new JobID(DUMMY_JOBTRACKER_ID, new Random().nextInt(Integer.MAX_VALUE))); LOG.info(MessageFormat.format( "starting job using {0}: {1} ({2})", this, job.getJobID(), job.getJobName())); try { runJob(job); return true; } catch (InterruptedException e) { throw e; } catch (Exception e) { LOG.error(MessageFormat.format( "exception was occurred while executing job: {0} ({1})", job.getJobID(), job.getJobName()), e); return false; } }
SimpleJobRunner implements JobRunner { @Override public boolean run(Job job) throws InterruptedException { job.setJobID(new JobID(DUMMY_JOBTRACKER_ID, new Random().nextInt(Integer.MAX_VALUE))); LOG.info(MessageFormat.format( "starting job using {0}: {1} ({2})", this, job.getJobID(), job.getJobName())); try { runJob(job); return true; } catch (InterruptedException e) { throw e; } catch (Exception e) { LOG.error(MessageFormat.format( "exception was occurred while executing job: {0} ({1})", job.getJobID(), job.getJobName()), e); return false; } } }
SimpleJobRunner implements JobRunner { @Override public boolean run(Job job) throws InterruptedException { job.setJobID(new JobID(DUMMY_JOBTRACKER_ID, new Random().nextInt(Integer.MAX_VALUE))); LOG.info(MessageFormat.format( "starting job using {0}: {1} ({2})", this, job.getJobID(), job.getJobName())); try { runJob(job); return true; } catch (InterruptedException e) { throw e; } catch (Exception e) { LOG.error(MessageFormat.format( "exception was occurred while executing job: {0} ({1})", job.getJobID(), job.getJobName()), e); return false; } }  }
SimpleJobRunner implements JobRunner { @Override public boolean run(Job job) throws InterruptedException { job.setJobID(new JobID(DUMMY_JOBTRACKER_ID, new Random().nextInt(Integer.MAX_VALUE))); LOG.info(MessageFormat.format( "starting job using {0}: {1} ({2})", this, job.getJobID(), job.getJobName())); try { runJob(job); return true; } catch (InterruptedException e) { throw e; } catch (Exception e) { LOG.error(MessageFormat.format( "exception was occurred while executing job: {0} ({1})", job.getJobID(), job.getJobName()), e); return false; } }  @Override boolean run(Job job); @Override String toString(); }
SimpleJobRunner implements JobRunner { @Override public boolean run(Job job) throws InterruptedException { job.setJobID(new JobID(DUMMY_JOBTRACKER_ID, new Random().nextInt(Integer.MAX_VALUE))); LOG.info(MessageFormat.format( "starting job using {0}: {1} ({2})", this, job.getJobID(), job.getJobName())); try { runJob(job); return true; } catch (InterruptedException e) { throw e; } catch (Exception e) { LOG.error(MessageFormat.format( "exception was occurred while executing job: {0} ({1})", job.getJobID(), job.getJobName()), e); return false; } }  @Override boolean run(Job job); @Override String toString(); static final String KEY_BUFFER_SIZE; static final String KEY_TEMPORARY_LOCATION; static final String KEY_COMPRESS_BLOCK; }