def _process_datum(self, data, input_reader, ctx, transient_shard_state):
    """Process a single data piece.

    Call mapper handler on the data.

    Args:
      data: a datum to process.
      input_reader: input reader.
      ctx: mapreduce context
      transient_shard_state: transient shard state.

    Returns:
      True if scan should be continued, False if scan should be stopped.
    """
    if data is not input_readers.ALLOW_CHECKPOINT:
      self.slice_context.incr(context.COUNTER_MAPPER_CALLS)

      handler = transient_shard_state.handler

      if isinstance(handler, map_job.Mapper):
        handler(self.slice_context, data)
      else:
        if input_reader.expand_parameters:
          result = handler(*data)
        else:
          result = handler(data)

        if util.is_generator(result):
          for output in result:
            if isinstance(output, operation.Operation):
              output(ctx)
            else:
              output_writer = transient_shard_state.output_writer
              if not output_writer:
                logging.warning(
                    "Handler yielded %s, but no output writer is set.", output)
              else:
                output_writer.write(output)

    if self._time() - self._start_time >= parameters.config._SLICE_DURATION_SEC:
      return False
    return True
def _process_datum ( self , data , input_reader , ctx , transient_shard_state ) : if data is not input_readers . ALLOW_CHECKPOINT : self . slice_context . incr ( context . COUNTER_MAPPER_CALLS ) handler = transient_shard_state . handler if isinstance ( handler , map_job . Mapper ) : handler ( self . slice_context , data ) else : if input_reader . expand_parameters : result = handler ( * data ) else : result = handler ( data ) if util . is_generator ( result ) : for output in result : if isinstance ( output , operation . Operation ) : output ( ctx ) else : output_writer = transient_shard_state . output_writer if not output_writer : logging . warning ( "Handler yielded %s, but no output writer is set." , output ) else : output_writer . write ( output ) if self . _time ( ) - self . _start_time >= parameters . config . _SLICE_DURATION_SEC : return False return True
Process a single data piece.

    Call mapper handler on the data.

    Args:
      data: a datum to process.
      input_reader: input reader.
      ctx: mapreduce context
      transient_shard_state: transient shard state.

    Returns:
      True if scan should be continued, False if scan should be stopped.