<s> """<STR_LIT>""" <EOL> import json <EOL> import re <EOL> from logging import getLogger <EOL> from mrjob . py2 import integer_types <EOL> from mrjob . py2 import string_types <EOL> from . counters import _sum_counters <EOL> from . ids import _add_implied_task_id <EOL> from . wrap import _ls_logs <EOL> from . wrap import _cat_log <EOL> log = getLogger ( __name__ ) <EOL> _HISTORY_LOG_PATH_RE = re . compile ( <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' ) <EOL> _PRE_YARN_HISTORY_ESCAPE_RE = re . compile ( r'<STR_LIT>' ) <EOL> _PRE_YARN_HISTORY_KEY_PAIR = re . compile ( <EOL> r'<STR_LIT>' , re . MULTILINE ) <EOL> _PRE_YARN_HISTORY_RECORD = re . compile ( <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' + _PRE_YARN_HISTORY_KEY_PAIR . pattern + '<STR_LIT>' <EOL> r'<STR_LIT>' , re . MULTILINE ) <EOL> _PRE_YARN_COUNTER_GROUP_RE = re . compile ( <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT:}>' ) <EOL> _PRE_YARN_COUNTER_RE = re . compile ( <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' ) <EOL> def _ls_history_logs ( fs , log_dir_stream , job_id = None ) : <EOL> """<STR_LIT>""" <EOL> return _ls_logs ( fs , log_dir_stream , _match_history_log_path , <EOL> job_id = job_id ) <EOL> def _match_history_log_path ( path , job_id = None ) : <EOL> """<STR_LIT>""" <EOL> m = _HISTORY_LOG_PATH_RE . match ( path ) <EOL> if not m : <EOL> return None <EOL> if not ( job_id is None or m . group ( '<STR_LIT>' ) == job_id ) : <EOL> return None <EOL> return dict ( job_id = m . group ( '<STR_LIT>' ) , yarn = '<STR_LIT>' in m . group ( '<STR_LIT>' ) ) <EOL> def _interpret_history_log ( fs , matches ) : <EOL> """<STR_LIT>""" <EOL> for match in matches : <EOL> path = match [ '<STR_LIT:path>' ] <EOL> if match [ '<STR_LIT>' ] : <EOL> result = _parse_yarn_history_log ( _cat_log ( fs , path ) ) <EOL> else : <EOL> result = _parse_pre_yarn_history_log ( _cat_log ( fs , path ) ) <EOL> for error in result . get ( '<STR_LIT>' ) or ( ) : <EOL> if '<STR_LIT>' in error : <EOL> error [ '<STR_LIT>' ] [ '<STR_LIT:path>' ] = path <EOL> _add_implied_task_id ( error ) <EOL> return result <EOL> return { } <EOL> def _parse_yarn_history_log ( lines ) : <EOL> """<STR_LIT>""" <EOL> result = { } <EOL> task_to_counters = { } <EOL> for line_num , line in enumerate ( lines ) : <EOL> if not line . startswith ( '<STR_LIT:{>' ) : <EOL> continue <EOL> try : <EOL> record = json . loads ( line ) <EOL> except : <EOL> continue <EOL> record_type = record . get ( '<STR_LIT:type>' ) <EOL> if not isinstance ( record_type , string_types ) : <EOL> continue <EOL> event_record = record . get ( '<STR_LIT>' ) <EOL> if not isinstance ( event_record , dict ) : <EOL> continue <EOL> events = [ e for e in record [ '<STR_LIT>' ] . values ( ) <EOL> if isinstance ( e , dict ) ] <EOL> for event in events : <EOL> if '<STR_LIT>' in event and '<STR_LIT>' in event : <EOL> result . setdefault ( '<STR_LIT>' , { } ) <EOL> result [ '<STR_LIT>' ] [ <EOL> event [ '<STR_LIT>' ] ] = event [ '<STR_LIT>' ] <EOL> if record_type . endswith ( '<STR_LIT>' ) : <EOL> for event in events : <EOL> err_msg = event . get ( '<STR_LIT:error>' ) <EOL> if not ( err_msg and isinstance ( err_msg , string_types ) ) : <EOL> continue <EOL> error = dict ( <EOL> hadoop_error = dict ( <EOL> message = err_msg , <EOL> start_line = line_num , <EOL> num_lines = <NUM_LIT:1> ) ) <EOL> if isinstance ( event . get ( '<STR_LIT>' ) , string_types ) : <EOL> error [ '<STR_LIT>' ] = event [ '<STR_LIT>' ] <EOL> if isinstance ( event . get ( '<STR_LIT>' ) , string_types ) : <EOL> error [ '<STR_LIT>' ] = event [ '<STR_LIT>' ] <EOL> result . setdefault ( '<STR_LIT>' , [ ] ) <EOL> result [ '<STR_LIT>' ] . append ( error ) <EOL> elif record_type == '<STR_LIT>' : <EOL> for event in events : <EOL> task_id = event . get ( '<STR_LIT>' ) <EOL> if not isinstance ( task_id , string_types ) : <EOL> continue <EOL> counters_record = event . get ( '<STR_LIT>' ) <EOL> if not isinstance ( counters_record , dict ) : <EOL> continue <EOL> task_to_counters [ task_id ] = _extract_yarn_counters ( <EOL> counters_record ) <EOL> elif record_type == '<STR_LIT>' : <EOL> for event in events : <EOL> counters_record = event . get ( '<STR_LIT>' ) <EOL> if not isinstance ( counters_record , dict ) : <EOL> continue <EOL> result [ '<STR_LIT>' ] = _extract_yarn_counters ( counters_record ) <EOL> if '<STR_LIT>' not in result and task_to_counters : <EOL> result [ '<STR_LIT>' ] = _sum_counters ( * task_to_counters . values ( ) ) <EOL> return result <EOL> def _extract_yarn_counters ( counters_record ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( counters_record , dict ) : <EOL> return { } <EOL> group_records = counters_record . get ( '<STR_LIT>' ) <EOL> if not isinstance ( group_records , list ) : <EOL> return { } <EOL> counters = { } <EOL> for group_record in group_records : <EOL> if not isinstance ( group_record , dict ) : <EOL> continue <EOL> group = group_record . get ( '<STR_LIT>' ) <EOL> if not isinstance ( group , string_types ) : <EOL> continue <EOL> counter_records = group_record . get ( '<STR_LIT>' ) <EOL> if not isinstance ( counter_records , list ) : <EOL> continue <EOL> for counter_record in counter_records : <EOL> counter = counter_record . get ( '<STR_LIT>' ) <EOL> if not isinstance ( counter , string_types ) : <EOL> continue <EOL> amount = counter_record . get ( '<STR_LIT:value>' ) <EOL> if not ( isinstance ( amount , integer_types ) ) : <EOL> continue <EOL> counters . setdefault ( group , { } ) <EOL> counters [ group ] . setdefault ( counter , <NUM_LIT:0> ) <EOL> counters [ group ] [ counter ] += amount <EOL> return counters <EOL> def _parse_pre_yarn_history_log ( lines ) : <EOL> """<STR_LIT>""" <EOL> result = { } <EOL> task_to_counters = { } <EOL> for record in _parse_pre_yarn_history_records ( lines ) : <EOL> fields = record [ '<STR_LIT>' ] <EOL> if record [ '<STR_LIT:type>' ] == '<STR_LIT>' and '<STR_LIT>' in fields : <EOL> result [ '<STR_LIT>' ] = _parse_pre_yarn_counters ( fields [ '<STR_LIT>' ] ) <EOL> elif ( record [ '<STR_LIT:type>' ] == '<STR_LIT>' and <EOL> '<STR_LIT>' in fields and '<STR_LIT>' in fields ) : <EOL> task_id = fields [ '<STR_LIT>' ] <EOL> counters = _parse_pre_yarn_counters ( fields [ '<STR_LIT>' ] ) <EOL> task_to_counters [ task_id ] = counters <EOL> elif ( record [ '<STR_LIT:type>' ] in ( '<STR_LIT>' , '<STR_LIT>' ) and <EOL> '<STR_LIT>' in fields and <EOL> fields . get ( '<STR_LIT>' ) == '<STR_LIT>' and <EOL> fields . get ( '<STR_LIT>' ) ) : <EOL> result . setdefault ( '<STR_LIT>' , [ ] ) <EOL> result [ '<STR_LIT>' ] . append ( dict ( <EOL> hadoop_error = dict ( <EOL> message = fields [ '<STR_LIT>' ] , <EOL> start_line = record [ '<STR_LIT>' ] , <EOL> num_lines = record [ '<STR_LIT>' ] ) , <EOL> attempt_id = fields [ '<STR_LIT>' ] ) ) <EOL> if '<STR_LIT>' not in result and task_to_counters : <EOL> result [ '<STR_LIT>' ] = _sum_counters ( * task_to_counters . values ( ) ) <EOL> return result <EOL> def _parse_pre_yarn_history_records ( lines ) : <EOL> """<STR_LIT>""" <EOL> def yield_record_strings ( lines ) : <EOL> record_lines = [ ] <EOL> start_line = <NUM_LIT:0> <EOL> for line_num , line in enumerate ( lines ) : <EOL> record_lines . append ( line ) <EOL> if line . endswith ( '<STR_LIT>' ) : <EOL> yield start_line , len ( record_lines ) , '<STR_LIT>' . join ( record_lines ) <EOL> record_lines = [ ] <EOL> start_line = line_num + <NUM_LIT:1> <EOL> for start_line , num_lines , record_str in yield_record_strings ( lines ) : <EOL> record_match = _PRE_YARN_HISTORY_RECORD . match ( record_str ) <EOL> if not record_match : <EOL> continue <EOL> record_type = record_match . group ( '<STR_LIT:type>' ) <EOL> key_pairs = record_match . group ( '<STR_LIT>' ) <EOL> fields = { } <EOL> for m in _PRE_YARN_HISTORY_KEY_PAIR . finditer ( key_pairs ) : <EOL> key = m . group ( '<STR_LIT:key>' ) <EOL> value = _pre_yarn_history_unescape ( m . group ( '<STR_LIT>' ) ) <EOL> fields [ key ] = value <EOL> yield dict ( <EOL> fields = fields , <EOL> num_lines = num_lines , <EOL> start_line = start_line , <EOL> type = record_type , <EOL> ) <EOL> def _parse_pre_yarn_counters ( counters_str ) : <EOL> """<STR_LIT>""" <EOL> counters = { } <EOL> for group_match in _PRE_YARN_COUNTER_GROUP_RE . finditer ( counters_str ) : <EOL> group_name = _pre_yarn_history_unescape ( <EOL> group_match . group ( '<STR_LIT>' ) ) <EOL> group_counters = { } <EOL> for counter_match in _PRE_YARN_COUNTER_RE . finditer ( <EOL> group_match . group ( '<STR_LIT>' ) ) : <EOL> counter_name = _pre_yarn_history_unescape ( <EOL> counter_match . group ( '<STR_LIT>' ) ) <EOL> amount = int ( counter_match . group ( '<STR_LIT>' ) ) <EOL> group_counters [ counter_name ] = amount <EOL> counters [ group_name ] = group_counters <EOL> return counters <EOL> def _pre_yarn_history_unescape ( s ) : <EOL> """<STR_LIT>""" <EOL> return _PRE_YARN_HISTORY_ESCAPE_RE . sub ( r'<STR_LIT>' , s ) </s>
