code: def combineByKey(self, createCombiner, mergeValue, mergeCombiners,
                       numPartitions=None, partitionFunc=portable_hash):
          if numPartitions is None:
              numPartitions = self._defaultReducePartitions()
          serializer = self.ctx.serializer
          memory = self._memory_limit()
          agg = Aggregator(createCombiner, mergeValue, mergeCombiners)
          def combineLocally(iterator):
              merger = ExternalMerger(agg, memory * 0.9, serializer)
              merger.mergeValues(iterator)
              return merger.items()
          locally_combined = self.mapPartitions(combineLocally, preservesPartitioning=True)
          shuffled = locally_combined.partitionBy(numPartitions, partitionFunc)
          def _mergeCombiners(iterator):
              merger = ExternalMerger(agg, memory, serializer)
              merger.mergeCombiners(iterator)
              return merger.items()
          return shuffled.mapPartitions(_mergeCombiners, preservesPartitioning=True)