code: def _reduce_input(self, inputs, reducer, final=NotImplemented):
          
          for key, values in groupby(inputs, key=lambda x: self.internal_serialize(x[0])):
              for output in reducer(self.deserialize(key), (v[1] for v in values)):
                  yield output
          if final != NotImplemented:
              for output in final():
                  yield output
          self._flush_batch_incr_counter()
docs: Iterate over input, collect values with the same key, and call the reducer for each unique key.