public void reduce ( K2 key , Iterator < V2 > values , OutputCollector < K3 , V3 > output , Reporter reporter ) throws IOException { isOk = false ; startApplication ( output , reporter ) ; downlink . reduceKey ( key ) ; while ( values . hasNext ( ) ) { downlink . reduceValue ( values . next ( ) ) ; } 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 ( ) ; } isOk = true ; }
Process all of the keys and values . Start up the application if we haven t started it yet .