#NEWLINE# def _run( self ) : #NEWLINE# #INDENT# self . _warn_ignored_opts ( ) #NEWLINE# _error_on_bad_paths ( self . fs , self . _input_paths ) #NEWLINE# self . _create_setup_wrapper_script ( local = True ) #NEWLINE# self . _setup_output_dir ( ) #NEWLINE# for ( step_num , step ) in enumerate ( self . _get_steps ( ) ) : #NEWLINE# #INDENT# log . info ( ( 'Running step %d of %d...' % ( ( step_num + 1 ) , self . _num_steps ( ) ) ) ) #NEWLINE# self . _check_step_works_with_runner ( step ) #NEWLINE# self . _counters . append ( { #NEWLINE# #INDENT# #NEWLINE# #UNINDENT# } ) #NEWLINE# self . _invoke_step ( step_num , 'mapper' ) #NEWLINE# if ( 'reducer' in step ) : #NEWLINE# #INDENT# sort_output_path = os . path . join ( self . _get_local_tmp_dir ( ) , ( 'step-%04d-mapper-sorted' % step_num ) ) #NEWLINE# self . _invoke_sort ( self . _step_input_paths ( ) , sort_output_path ) #NEWLINE# self . _prev_outfiles = [ sort_output_path ] #NEWLINE# self . _invoke_step ( step_num , 'reducer' ) #NEWLINE# #UNINDENT# #UNINDENT# for ( i , outfile ) in enumerate ( self . _prev_outfiles ) : #NEWLINE# #INDENT# final_outfile = os . path . join ( step . _output_dir , ( 'part-%05d' % i ) ) #NEWLINE# log . debug ( ( 'Moving %s -> %s' % ( outfile , final_outfile ) ) ) #NEWLINE# shutil . move ( outfile , final_outfile )