@SuppressWarnings("unchecked")
    public <INKEY,INVALUE,OUTKEY,OUTVALUE,
        T extends org.apache.hadoop.mapreduce.InputSplit> 
    void run() throws Exception {
        Configuration conf = job.getConfiguration();
        reporter = new ContentPumpReporter();
        InputFormat<INKEY,INVALUE> inputFormat = 
            (InputFormat<INKEY, INVALUE>)ReflectionUtils.newInstance(
                job.getInputFormatClass(), conf);
        List<InputSplit> splits;
        T[] array;
        try {
            splits = inputFormat.getSplits(job);
            array = (T[])splits.toArray(
                    new org.apache.hadoop.mapreduce.InputSplit[splits.size()]);
            // sort the splits into order based on size, so that the biggest
            // goes first
            Arrays.sort(array, new SplitLengthComparator());
        } catch (Exception ex) {
            if (LOG.isDebugEnabled()) {
                LOG.debug("Error getting input splits: ", ex);
            } else {
                LOG.error("Error getting input splits: ");
                LOG.error(ex.getMessage());
            }
            job.setJobState(JobStatus.State.FAILED);
            return;
        }               
        OutputFormat<OUTKEY, OUTVALUE> outputFormat = 
            (OutputFormat<OUTKEY, OUTVALUE>)ReflectionUtils.newInstance(
                job.getOutputFormatClass(), conf);
        Class<? extends Mapper<?,?,?,?>> mapperClass = job.getMapperClass();
        Mapper<INKEY,INVALUE,OUTKEY,OUTVALUE> mapper = 
            (Mapper<INKEY,INVALUE,OUTKEY,OUTVALUE>)ReflectionUtils.newInstance(
                mapperClass, conf);
        try {
            outputFormat.checkOutputSpecs(job);
        } catch (Exception ex) {         
            if (LOG.isDebugEnabled()) {
                LOG.debug("Error checking output specification: ", ex);
            } else {
                LOG.error("Error checking output specification: ");
                LOG.error(ex.getMessage());
            }
            job.setJobState(JobStatus.State.FAILED);
            return;
        }
        progress = new AtomicInteger[splits.size()];
        for (int i = 0; i < splits.size(); i++) {
            progress[i] = new AtomicInteger();
        }
     
        job.setJobState(JobStatus.State.RUNNING);
        Monitor monitor = new Monitor();
        monitor.start();
        List<Future<Object>> taskList = new ArrayList<Future<Object>>();
        for (int i = 0; i < array.length && !ContentPump.shutdown; i++) {        
            InputSplit split = array[i];
            if (pool != null) {
                LocalMapTask<INKEY, INVALUE, OUTKEY, OUTVALUE> task = 
                    new LocalMapTask<INKEY, INVALUE, OUTKEY, OUTVALUE>(
                        inputFormat, outputFormat, conf, i, split, reporter,
                        progress[i]);
                availableThreads = assignThreads(i, array.length);
                Class<? extends Mapper<?, ?, ?, ?>> runtimeMapperClass = 
                    job.getMapperClass();
                if (availableThreads > 1 && 
                    availableThreads != threadsPerSplit) { 
                	// possible runtime adjustment
                    if (runtimeMapperClass != (Class)MultithreadedMapper.class) {
                	    runtimeMapperClass = (Class<? extends 
                	        Mapper<INKEY, INVALUE, OUTKEY, OUTVALUE>>)
                	        cmd.getRuntimeMapperClass(job, mapperClass, 
                		        threadsPerSplit, availableThreads);
                    }   
                    if (runtimeMapperClass != mapperClass) {
                	    task.setMapperClass(runtimeMapperClass);
                    }
                    if (runtimeMapperClass == (Class)MultithreadedMapper.class) {
                	    task.setThreadCount(availableThreads);
                	    if (LOG.isDebugEnabled()) {
                            LOG.debug("Thread Count for Split#" + i + " : "
                                    + availableThreads);
                        }
                    }
                }
                
                if (runtimeMapperClass == (Class)MultithreadedMapper.class) {
                    synchronized (pool) {
                        taskList.add(pool.submit(task));
                        pool.wait();
                    }
                } else {
                	pool.submit(task);
                }
            } else { // single-threaded
                JobID jid = new JobID();
                TaskID taskId = new TaskID(jid.getJtIdentifier(), jid.getId(), TaskType.MAP, i);
                TaskAttemptID taskAttemptId = new TaskAttemptID(taskId, 0);
                TaskAttemptContext context = 
                    ReflectionUtil.createTaskAttemptContext(conf, taskAttemptId);
                RecordReader<INKEY, INVALUE> reader = 
                    inputFormat.createRecordReader(split, context);
                RecordWriter<OUTKEY, OUTVALUE> writer = 
                    outputFormat.getRecordWriter(context);
                OutputCommitter committer = 
                    outputFormat.getOutputCommitter(context);
                TrackingRecordReader trackingReader = 
                    new TrackingRecordReader(reader, progress[i]);

                Mapper.Context mapperContext = 
                    ReflectionUtil.createMapperContext(mapper, conf, 
                        taskAttemptId, trackingReader, writer, committer, 
                        reporter, split);
                
                trackingReader.initialize(split, mapperContext);
                
                // no thread pool (only 1 thread specified)
                Class<? extends Mapper<?,?,?,?>> mapClass = 
                        job.getMapperClass();
                mapperContext.getConfiguration().setClass(
                   CONF_MAPREDUCE_JOB_MAP_CLASS , mapClass, Mapper.class);
                mapper = (Mapper<INKEY, INVALUE, OUTKEY, OUTVALUE>) 
                    ReflectionUtils.newInstance(mapClass,
                        mapperContext.getConfiguration());
                try {
                    mapper.run(mapperContext);
                } finally {
                    try {
                        trackingReader.close();
                    } catch (Throwable t) {
                        LOG.error("Error closing reader: " + t.getMessage());
                        if (LOG.isDebugEnabled()) {
                            LOG.debug(t);
                        }
                    }
                    try {
                        writer.close(mapperContext);
                    } catch (Throwable t) {
                        LOG.error("Error closing writer: " + t.getMessage());
                        if (LOG.isDebugEnabled()) {
                            LOG.debug(t);
                        }
                    } 
                    try {
                        committer.commitTask(context);
                    } catch (Throwable t) {
                        LOG.error("Error committing task: " + t.getMessage());
                        if (LOG.isDebugEnabled()) {
                            LOG.debug(t);
                        }
                    }
                }
            }
        }
        // wait till all tasks are done
        if (pool != null) {
            pool.shutdown();
            while (!pool.awaitTermination(1, TimeUnit.HOURS));
        } 
        job.setJobState(JobStatus.State.SUCCEEDED);
        monitor.interrupt();
        monitor.join(1000);
        
        // report counters
        Iterator<CounterGroup> groupIt = 
            reporter.counters.iterator();
        while (groupIt.hasNext()) {
            CounterGroup group = groupIt.next();
            LOG.info(group.getDisplayName() + ": ");
            Iterator<Counter> counterIt = group.iterator();
            while (counterIt.hasNext()) {
                Counter counter = counterIt.next();
                LOG.info(counter.getDisplayName() + ": " + 
                                counter.getValue());
            }
        }
        LOG.info("Total execution time: " + 
                 (System.currentTimeMillis() - startTime) / 1000 + " sec");
    }
@ SuppressWarnings ( "unchecked" ) public < INKEY , INVALUE , OUTKEY , OUTVALUE , T extends org . apache . hadoop . mapreduce . InputSplit > void run ( ) throws Exception { Configuration conf = job . getConfiguration ( ) ; reporter = new ContentPumpReporter ( ) ; InputFormat < INKEY , INVALUE > inputFormat = ( InputFormat < INKEY , INVALUE > ) ReflectionUtils . newInstance ( job . getInputFormatClass ( ) , conf ) ; List < InputSplit > splits ; T [ ] array ; try { splits = inputFormat . getSplits ( job ) ; array = ( T [ ] ) splits . toArray ( new org . apache . hadoop . mapreduce . InputSplit [ splits . size ( ) ] ) ; Arrays . sort ( array , new SplitLengthComparator ( ) ) ; } catch ( Exception ex ) { if ( LOG . isDebugEnabled ( ) ) { LOG . debug ( "Error getting input splits: " , ex ) ; } else { LOG . error ( "Error getting input splits: " ) ; LOG . error ( ex . getMessage ( ) ) ; } job . setJobState ( JobStatus . State . FAILED ) ; return ; } OutputFormat < OUTKEY , OUTVALUE > outputFormat = ( OutputFormat < OUTKEY , OUTVALUE > ) ReflectionUtils . newInstance ( job . getOutputFormatClass ( ) , conf ) ; Class < ? extends Mapper < ? , ? , ? , ? > > mapperClass = job . getMapperClass ( ) ; Mapper < INKEY , INVALUE , OUTKEY , OUTVALUE > mapper = ( Mapper < INKEY , INVALUE , OUTKEY , OUTVALUE > ) ReflectionUtils . newInstance ( mapperClass , conf ) ; try { outputFormat . checkOutputSpecs ( job ) ; } catch ( Exception ex ) { if ( LOG . isDebugEnabled ( ) ) { LOG . debug ( "Error checking output specification: " , ex ) ; } else { LOG . error ( "Error checking output specification: " ) ; LOG . error ( ex . getMessage ( ) ) ; } job . setJobState ( JobStatus . State . FAILED ) ; return ; } progress = new AtomicInteger [ splits . size ( ) ] ; for ( int i = 0 ; i < splits . size ( ) ; i ++ ) { progress [ i ] = new AtomicInteger ( ) ; } job . setJobState ( JobStatus . State . RUNNING ) ; Monitor monitor = new Monitor ( ) ; monitor . start ( ) ; List < Future < Object > > taskList = new ArrayList < Future < Object > > ( ) ; for ( int i = 0 ; i < array . length && ! ContentPump . shutdown ; i ++ ) { InputSplit split = array [ i ] ; if ( pool != null ) { LocalMapTask < INKEY , INVALUE , OUTKEY , OUTVALUE > task = new LocalMapTask < INKEY , INVALUE , OUTKEY , OUTVALUE > ( inputFormat , outputFormat , conf , i , split , reporter , progress [ i ] ) ; availableThreads = assignThreads ( i , array . length ) ; Class < ? extends Mapper < ? , ? , ? , ? > > runtimeMapperClass = job . getMapperClass ( ) ; if ( availableThreads > 1 && availableThreads != threadsPerSplit ) { if ( runtimeMapperClass != ( Class ) MultithreadedMapper . class ) { runtimeMapperClass = ( Class < ? extends Mapper < INKEY , INVALUE , OUTKEY , OUTVALUE > > ) cmd . getRuntimeMapperClass ( job , mapperClass , threadsPerSplit , availableThreads ) ; } if ( runtimeMapperClass != mapperClass ) { task . setMapperClass ( runtimeMapperClass ) ; } if ( runtimeMapperClass == ( Class ) MultithreadedMapper . class ) { task . setThreadCount ( availableThreads ) ; if ( LOG . isDebugEnabled ( ) ) { LOG . debug ( "Thread Count for Split#" + i + " : " + availableThreads ) ; } } } if ( runtimeMapperClass == ( Class ) MultithreadedMapper . class ) { synchronized ( pool ) { taskList . add ( pool . submit ( task ) ) ; pool . wait ( ) ; } } else { pool . submit ( task ) ; } } else { JobID jid = new JobID ( ) ; TaskID taskId = new TaskID ( jid . getJtIdentifier ( ) , jid . getId ( ) , TaskType . MAP , i ) ; TaskAttemptID taskAttemptId = new TaskAttemptID ( taskId , 0 ) ; TaskAttemptContext context = ReflectionUtil . createTaskAttemptContext ( conf , taskAttemptId ) ; RecordReader < INKEY , INVALUE > reader = inputFormat . createRecordReader ( split , context ) ; RecordWriter < OUTKEY , OUTVALUE > writer = outputFormat . getRecordWriter ( context ) ; OutputCommitter committer = outputFormat . getOutputCommitter ( context ) ; TrackingRecordReader trackingReader = new TrackingRecordReader ( reader , progress [ i ] ) ; Mapper . Context mapperContext = ReflectionUtil . createMapperContext ( mapper , conf , taskAttemptId , trackingReader , writer , committer , reporter , split ) ; trackingReader . initialize ( split , mapperContext ) ; Class < ? extends Mapper < ? , ? , ? , ? > > mapClass = job . getMapperClass ( ) ; mapperContext . getConfiguration ( ) . setClass ( CONF_MAPREDUCE_JOB_MAP_CLASS , mapClass , Mapper . class ) ; mapper = ( Mapper < INKEY , INVALUE , OUTKEY , OUTVALUE > ) ReflectionUtils . newInstance ( mapClass , mapperContext . getConfiguration ( ) ) ; try { mapper . run ( mapperContext ) ; } finally { try { trackingReader . close ( ) ; } catch ( Throwable t ) { LOG . error ( "Error closing reader: " + t . getMessage ( ) ) ; if ( LOG . isDebugEnabled ( ) ) { LOG . debug ( t ) ; } } try { writer . close ( mapperContext ) ; } catch ( Throwable t ) { LOG . error ( "Error closing writer: " + t . getMessage ( ) ) ; if ( LOG . isDebugEnabled ( ) ) { LOG . debug ( t ) ; } } try { committer . commitTask ( context ) ; } catch ( Throwable t ) { LOG . error ( "Error committing task: " + t . getMessage ( ) ) ; if ( LOG . isDebugEnabled ( ) ) { LOG . debug ( t ) ; } } } } } if ( pool != null ) { pool . shutdown ( ) ; while ( ! pool . awaitTermination ( 1 , TimeUnit . HOURS ) ) ; } job . setJobState ( JobStatus . State . SUCCEEDED ) ; monitor . interrupt ( ) ; monitor . join ( 1000 ) ; Iterator < CounterGroup > groupIt = reporter . counters . iterator ( ) ; while ( groupIt . hasNext ( ) ) { CounterGroup group = groupIt . next ( ) ; LOG . info ( group . getDisplayName ( ) + ": " ) ; Iterator < Counter > counterIt = group . iterator ( ) ; while ( counterIt . hasNext ( ) ) { Counter counter = counterIt . next ( ) ; LOG . info ( counter . getDisplayName ( ) + ": " + counter . getValue ( ) ) ; } } LOG . info ( "Total execution time: " + ( System . currentTimeMillis ( ) - startTime ) / 1000 + " sec" ) ; }
Run the job.  Get the input splits, create map tasks and submit it to
the thread pool if there is one; otherwise, runs the the task one by
one.

@param <INKEY>
@param <INVALUE>
@param <OUTKEY>
@param <OUTVALUE>
@throws Exception