def process_entity(self, entity, ctx):
    """Process a single entity.

    Call mapper handler on the entity.

    Args:
      entity: an entity to process.
      ctx: current execution context.

    Returns:
      True if scan should be continued, False if scan should be aborted.
    """
    ctx.counters.increment(context.COUNTER_MAPPER_CALLS)

    handler = ctx.mapreduce_spec.mapper.handler
    if util.is_generator_function(handler):
      for result in handler(entity):
        if callable(result):
          result(ctx)
        else:
          try:
            if len(result) == 2:
              logging.error("Collectors not implemented yet")
            else:
              logging.error("Got bad output tuple of length %d", len(result))
          except TypeError:
            logging.error(
                "Handler yielded type %s, expected a callable or a tuple",
                result.__class__.__name__)
    else:
      handler(entity)

    if self._time() - self._start_time > _SLICE_DURATION_SEC:
      logging.debug("Spent %s seconds. Rescheduling",
                    self._time() - self._start_time)
      return False
    return True
Process a single entity.

    Call mapper handler on the entity.

    Args:
      entity: an entity to process.
      ctx: current execution context.

    Returns:
      True if scan should be continued, False if scan should be aborted.
    
mapreduce/handlers.py