@ SuppressWarnings ( "unchecked" ) public void run ( RecordReader < K1 , V1 > input , OutputCollector < K2 , V2 > output , Reporter reporter ) throws IOException { Application < K1 , V1 , K2 , V2 > application = null ; try { RecordReader < FloatWritable , NullWritable > fakeInput = ( ! Submitter . getIsJavaRecordReader ( job ) && ! Submitter . getIsJavaMapper ( job ) ) ? ( RecordReader < FloatWritable , NullWritable > ) input : null ; application = new Application < K1 , V1 , K2 , V2 > ( job , fakeInput , output , reporter , ( Class < ? extends K2 > ) job . getOutputKeyClass ( ) , ( Class < ? extends V2 > ) job . getOutputValueClass ( ) ) ; } catch ( InterruptedException ie ) { throw new RuntimeException ( "interrupted" , ie ) ; } DownwardProtocol < K1 , V1 > downlink = application . getDownlink ( ) ; boolean isJavaInput = Submitter . getIsJavaRecordReader ( job ) ; downlink . runMap ( reporter . getInputSplit ( ) , job . getNumReduceTasks ( ) , isJavaInput ) ; boolean skipping = job . getBoolean ( "mapred.skip.on" , false ) ; try { if ( isJavaInput ) { // allocate key & value instances that are re-used for all entries K1 key = input . createKey ( ) ; V1 value = input . createValue ( ) ; downlink . setInputTypes ( key . getClass ( ) . getName ( ) , value . getClass ( ) . getName ( ) ) ; while ( input . next ( key , value ) ) { // map pair to output downlink . mapItem ( key , value ) ; if ( skipping ) { //flush the streams on every record input if running in skip mode //so that we don't buffer other records surrounding a bad record. downlink . flush ( ) ; } } downlink . endOfInput ( ) ; } application . waitForFinish ( ) ; } catch ( Throwable t ) { application . abort ( t ) ; } finally { application . cleanup ( ) ; } }
Run the map task .