code: 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
docs: 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.