code: def reduceByKeyLocally(self, func):
          func = fail_on_stopiteration(func)
          def reducePartition(iterator):
              m = {}
              for k, v in iterator:
                  m[k] = func(m[k], v) if k in m else v
              yield m
          def mergeMaps(m1, m2):
              for k, v in m2.items():
                  m1[k] = func(m1[k], v) if k in m1 else v
              return m1
          return self.mapPartitions(reducePartition).reduce(mergeMaps)