code: def _map_input(self, input_stream):
          
          for record in self.reader(input_stream):
              for output in self.mapper(*record):
                  yield output
          if self.final_mapper != NotImplemented:
              for output in self.final_mapper():
                  yield output
          self._flush_batch_incr_counter()
docs: Iterate over input and call the mapper for each item.
          If the job has a parser defined, the return values from the parser will
          be passed as arguments to the mapper.
  
          If the input is coded output from a previous run,
          the arguments will be splitted in key and value.