<s> """<STR_LIT>""" <EOL> import logging <EOL> from subprocess import CalledProcessError <EOL> from subprocess import Popen <EOL> from subprocess import PIPE <EOL> from mrjob . logs . counters import _format_counters <EOL> from mrjob . parse import _find_python_traceback <EOL> from mrjob . parse import parse_mr_job_stderr <EOL> from mrjob . py2 import string_types <EOL> from mrjob . sim import SimMRJobRunner <EOL> from mrjob . step import StepFailedException <EOL> from mrjob . util import cmd_line <EOL> from mrjob . util import shlex_split <EOL> log = logging . getLogger ( __name__ ) <EOL> def _chain_procs ( procs_args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> last_stdout = None <EOL> procs = [ ] <EOL> for i , args in enumerate ( procs_args ) : <EOL> proc_kwargs = kwargs . copy ( ) <EOL> if i > <NUM_LIT:0> : <EOL> proc_kwargs [ '<STR_LIT>' ] = last_stdout <EOL> if i < len ( procs_args ) - <NUM_LIT:1> : <EOL> proc_kwargs [ '<STR_LIT>' ] = PIPE <EOL> proc = Popen ( args , ** proc_kwargs ) <EOL> last_stdout = proc . stdout <EOL> procs . append ( proc ) <EOL> return procs <EOL> class LocalMRJobRunner ( SimMRJobRunner ) : <EOL> """<STR_LIT>""" <EOL> alias = '<STR_LIT>' <EOL> def __init__ ( self , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> super ( LocalMRJobRunner , self ) . __init__ ( ** kwargs ) <EOL> self . _all_proc_dicts = [ ] <EOL> self . _internal_jobconf = { } <EOL> def _run_step ( self , step_num , step_type , input_path , output_path , <EOL> working_dir , env ) : <EOL> step = self . _get_step ( step_num ) <EOL> if step_type == '<STR_LIT>' : <EOL> procs_args = self . _mapper_arg_chain ( <EOL> step , step_num , input_path ) <EOL> elif step_type == '<STR_LIT>' : <EOL> procs_args = self . _reducer_arg_chain ( <EOL> step , step_num , input_path ) <EOL> proc_dicts = self . _invoke_processes ( <EOL> procs_args , output_path , working_dir , env ) <EOL> self . _all_proc_dicts . extend ( proc_dicts ) <EOL> def _per_step_runner_finish ( self , step_num ) : <EOL> for proc_dict in self . _all_proc_dicts : <EOL> self . _wait_for_process ( proc_dict , step_num ) <EOL> self . _all_proc_dicts = [ ] <EOL> def _filter_if_any ( self , substep_dict ) : <EOL> if substep_dict [ '<STR_LIT:type>' ] == '<STR_LIT>' : <EOL> if '<STR_LIT>' in substep_dict : <EOL> return shlex_split ( substep_dict [ '<STR_LIT>' ] ) <EOL> return None <EOL> def _substep_args ( self , step_dict , step_num , mrc , input_path = None ) : <EOL> if step_dict [ '<STR_LIT:type>' ] != '<STR_LIT>' : <EOL> raise Exception ( "<STR_LIT>" % <EOL> step_dict [ '<STR_LIT:type>' ] ) <EOL> if step_dict [ mrc ] [ '<STR_LIT:type>' ] == '<STR_LIT>' : <EOL> if input_path is None : <EOL> return [ shlex_split ( step_dict [ mrc ] [ '<STR_LIT>' ] ) ] <EOL> else : <EOL> return [ <EOL> [ '<STR_LIT>' , input_path ] , <EOL> shlex_split ( step_dict [ mrc ] [ '<STR_LIT>' ] ) ] <EOL> if step_dict [ mrc ] [ '<STR_LIT:type>' ] == '<STR_LIT>' : <EOL> args = self . _script_args_for_step ( step_num , mrc ) <EOL> if input_path is None : <EOL> return [ args ] <EOL> else : <EOL> return [ args + [ input_path ] ] <EOL> def _substep_arg_chain ( self , mrc , step_dict , step_num , input_path ) : <EOL> procs_args = [ ] <EOL> filter_args = self . _filter_if_any ( step_dict [ mrc ] ) <EOL> if filter_args : <EOL> procs_args . append ( [ '<STR_LIT>' , input_path ] ) <EOL> procs_args . append ( filter_args ) <EOL> procs_args . extend ( <EOL> self . _substep_args ( step_dict , step_num , mrc ) ) <EOL> else : <EOL> procs_args . extend ( <EOL> self . _substep_args ( step_dict , step_num , mrc , input_path ) ) <EOL> return procs_args <EOL> def _mapper_arg_chain ( self , step_dict , step_num , input_path ) : <EOL> if '<STR_LIT>' not in step_dict : <EOL> new_step_dict = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> } <EOL> new_step_dict . update ( step_dict ) <EOL> step_dict = new_step_dict <EOL> procs_args = self . _substep_arg_chain ( <EOL> '<STR_LIT>' , step_dict , step_num , input_path ) <EOL> if '<STR_LIT>' in step_dict : <EOL> procs_args . append ( [ '<STR_LIT>' ] ) <EOL> procs_args . extend ( self . _combiner_arg_chain ( step_dict , step_num ) ) <EOL> return procs_args <EOL> def _combiner_arg_chain ( self , step_dict , step_num ) : <EOL> procs_args = [ ] <EOL> filter_args = self . _filter_if_any ( step_dict [ '<STR_LIT>' ] ) <EOL> if filter_args : <EOL> procs_args . append ( filter_args ) <EOL> procs_args . extend ( <EOL> self . _substep_args ( step_dict , step_num , '<STR_LIT>' ) ) <EOL> return procs_args <EOL> def _reducer_arg_chain ( self , step_dict , step_num , input_path ) : <EOL> return self . _substep_arg_chain ( <EOL> '<STR_LIT>' , step_dict , step_num , input_path ) <EOL> def _invoke_processes ( self , procs_args , output_path , working_dir , env ) : <EOL> """<STR_LIT>""" <EOL> log . debug ( '<STR_LIT>' % ( '<STR_LIT>' . join ( <EOL> args if isinstance ( args , string_types ) else cmd_line ( args ) <EOL> for args in procs_args ) , output_path ) ) <EOL> with open ( output_path , '<STR_LIT:wb>' ) as write_to : <EOL> procs = _chain_procs ( procs_args , stdout = write_to , stderr = PIPE , <EOL> cwd = working_dir , env = env ) <EOL> return [ { '<STR_LIT:args>' : a , '<STR_LIT>' : proc , '<STR_LIT>' : write_to } <EOL> for a , proc in zip ( procs_args , procs ) ] <EOL> def _wait_for_process ( self , proc_dict , step_num ) : <EOL> proc = proc_dict [ '<STR_LIT>' ] <EOL> stderr_lines = self . _process_stderr_from_script ( <EOL> proc . stderr , step_num = step_num ) <EOL> tb_lines = _find_python_traceback ( stderr_lines ) <EOL> if proc . stdout : <EOL> proc . stdout . close ( ) <EOL> proc . stderr . close ( ) <EOL> returncode = proc . wait ( ) <EOL> if returncode != <NUM_LIT:0> : <EOL> counters = self . _counters [ step_num ] <EOL> if counters : <EOL> log . info ( _format_counters ( counters ) ) <EOL> if tb_lines : <EOL> for line in tb_lines : <EOL> log . error ( line . rstrip ( '<STR_LIT:\r\n>' ) ) <EOL> reason = str ( <EOL> CalledProcessError ( returncode , proc_dict [ '<STR_LIT:args>' ] ) ) <EOL> raise StepFailedException ( <EOL> reason = reason , step_num = step_num , <EOL> num_steps = len ( self . _get_steps ( ) ) ) <EOL> def _process_stderr_from_script ( self , stderr , step_num = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> for line in stderr : <EOL> parsed = parse_mr_job_stderr ( <EOL> [ line ] , counters = self . _counters [ step_num ] ) <EOL> for status in parsed [ '<STR_LIT>' ] : <EOL> log . info ( '<STR_LIT>' % status ) <EOL> for line in parsed [ '<STR_LIT>' ] : <EOL> log . debug ( '<STR_LIT>' % line . rstrip ( '<STR_LIT:\r\n>' ) ) <EOL> yield line <EOL> def _default_python_bin ( self , local = False ) : <EOL> return super ( LocalMRJobRunner , self ) . _default_python_bin ( <EOL> local = True ) </s>
