def _parse_yarn_history_log(lines):
    """Collect useful info from a YARN history file, dealing gracefully
    with unexpected data structures.

    This returns a dictionary which may contain the following keys:

    container_to_attempt_id: map from container_id to attempt_id (useful
        for matching with task errors where we don't know attempt ID)
    counters: map from group to counter to amount. If job failed, we sum
        counters for succesful tasks
    errors: a list of dictionaries with the keys:
        hadoop_error:
            message: lines of error, as as string
            start_line: first line of log containing the error (0-indexed)
            num_lines: # of lines of log containing the error
        task_id: ID of task with this error
        attempt_id: ID of task attempt with this error
    """
    result = {}
    task_to_counters = {}  # used for successful tasks in failed jobs

    for line_num, line in enumerate(lines):
        # empty space or "Avro-Json" header
        if not line.startswith('{'):
            continue

        try:
            record = json.loads(line)
        except:
            continue

        record_type = record.get('type')
        if not isinstance(record_type, string_types):
            continue

        # extract events. Looks like there's just one per record
        event_record = record.get('event')
        if not isinstance(event_record, dict):
            continue
        events = [e for e in record['event'].values()
                  if isinstance(e, dict)]

        # update container_id -> attempt_id mapping
        for event in events:
            if event in 'attemptId' and 'containerId' in event:
                result.setdefault('container_to_attempt_id', {})
                result['container_to_attempt_id'][
                    event['containerId']] = event['attemptId']

        if record_type.endswith('_ATTEMPT_FAILED'):
            for event in events:
                err_msg = event.get('error')
                if not (err_msg and isinstance(err_msg, string_types)):
                    continue

                error = dict(
                    hadoop_error=dict(
                        message=err_msg,
                        start_line=line_num,
                        num_lines=1))

                if isinstance(event.get('taskid'), string_types):
                    error['task_id'] = event['taskid']

                if isinstance(event.get('attemptId'), string_types):
                    error['attempt_id'] = event['attemptId']

                result.setdefault('errors', [])
                result['errors'].append(error)

        elif record_type == 'TASK_FINISHED':
            for event in events:
                task_id = event.get('taskid')
                if not isinstance(task_id, string_types):
                    continue

                counters_record = event.get('counters')
                if not isinstance(counters_record, dict):
                    continue

                task_to_counters[task_id] = _extract_yarn_counters(
                    counters_record)

        elif record_type == 'JOB_FINISHED':
            for event in events:
                # mapCounters and reduceCounters are also available
                counters_record = event.get('totalCounters')
                if not isinstance(counters_record, dict):
                    continue

                result['counters'] = _extract_yarn_counters(counters_record)

    # if job failed, patch together counters from successful tasks
    if 'counters' not in result and task_to_counters:
        result['counters'] = _sum_counters(*task_to_counters.values())

    return result
dataset/ETHPy150Open Yelp/mrjob/mrjob/logs/history.py/_parse_yarn_history_log/swapped operands of `in`