nodes: 0, module; 1, function_definition; 2, function_name:combineByKey; 3, parameters; 4, block; 5, identifier:self; 6, identifier:createCombiner; 7, identifier:mergeValue; 8, identifier:mergeCombiners; 9, default_parameter; 10, default_parameter; 11, expression_statement; 12, if_statement; 13, expression_statement; 14, expression_statement; 15, expression_statement; 16, function_definition; 17, expression_statement; 18, expression_statement; 19, function_definition; 20, return_statement; 21, identifier:numPartitions; 22, None; 23, identifier:partitionFunc; 24, identifier:portable_hash; 25, comment:"""
          Generic function to combine the elements for each key using a custom
          set of aggregation functions.
  
          Turns an RDD[(K, V)] into a result of type RDD[(K, C)], for a "combined
          type" C.
  
          Users provide three functions:
  
              - C{createCombiner}, which turns a V into a C (e.g., creates
                a one-element list)
              - C{mergeValue}, to merge a V into a C (e.g., adds it to the end of
                a list)
              - C{mergeCombiners}, to combine two C's into a single one (e.g., merges
                the lists)
  
          To avoid memory allocation, both mergeValue and mergeCombiners are allowed to
          modify and return their first argument instead of creating a new C.
  
          In addition, users can control the partitioning of the output RDD.
  
          .. note:: V and C can be different -- for example, one might group an RDD of type
              (Int, Int) into an RDD of type (Int, List[Int]).
  
          >>> x = sc.parallelize([("a", 1), ("b", 1), ("a", 2)])
          >>> def to_list(a):
          ...     return [a]
          ...
          >>> def append(a, b):
          ...     a.append(b)
          ...     return a
          ...
          >>> def extend(a, b):
          ...     a.extend(b)
          ...     return a
          ...
          >>> sorted(x.combineByKey(to_list, append, extend).collect())
          [('a', [1, 2]), ('b', [1])]
          """; 26, comparison_operator:numPartitions is None; 27, block; 28, assignment; 29, assignment; 30, assignment; 31, function_name:combineLocally; 32, parameters; 33, block; 34, assignment; 35, assignment; 36, function_name:_mergeCombiners; 37, parameters; 38, block; 39, call; 40, identifier:numPartitions; 41, None; 42, expression_statement; 43, identifier:serializer; 44, attribute; 45, identifier:memory; 46, call; 47, identifier:agg; 48, call; 49, identifier:iterator; 50, expression_statement; 51, expression_statement; 52, return_statement; 53, identifier:locally_combined; 54, call; 55, identifier:shuffled; 56, call; 57, identifier:iterator; 58, expression_statement; 59, expression_statement; 60, return_statement; 61, attribute; 62, argument_list; 63, assignment; 64, attribute; 65, identifier:serializer; 66, attribute; 67, argument_list; 68, identifier:Aggregator; 69, argument_list; 70, assignment; 71, call; 72, call; 73, attribute; 74, argument_list; 75, attribute; 76, argument_list; 77, assignment; 78, call; 79, call; 80, identifier:shuffled; 81, identifier:mapPartitions; 82, identifier:_mergeCombiners; 83, keyword_argument; 84, identifier:numPartitions; 85, call; 86, identifier:self; 87, identifier:ctx; 88, identifier:self; 89, identifier:_memory_limit; 90, identifier:createCombiner; 91, identifier:mergeValue; 92, identifier:mergeCombiners; 93, identifier:merger; 94, call; 95, attribute; 96, argument_list; 97, attribute; 98, argument_list; 99, identifier:self; 100, identifier:mapPartitions; 101, identifier:combineLocally; 102, keyword_argument; 103, identifier:locally_combined; 104, identifier:partitionBy; 105, identifier:numPartitions; 106, identifier:partitionFunc; 107, identifier:merger; 108, call; 109, attribute; 110, argument_list; 111, attribute; 112, argument_list; 113, identifier:preservesPartitioning; 114, True; 115, attribute; 116, argument_list; 117, identifier:ExternalMerger; 118, argument_list; 119, identifier:merger; 120, identifier:mergeValues; 121, identifier:iterator; 122, identifier:merger; 123, identifier:items; 124, identifier:preservesPartitioning; 125, True; 126, identifier:ExternalMerger; 127, argument_list; 128, identifier:merger; 129, identifier:mergeCombiners; 130, identifier:iterator; 131, identifier:merger; 132, identifier:items; 133, identifier:self; 134, identifier:_defaultReducePartitions; 135, identifier:agg; 136, binary_operator:memory * 0.9; 137, identifier:serializer; 138, identifier:agg; 139, identifier:memory; 140, identifier:serializer; 141, identifier:memory; 142, float:0.9
code: def combineByKey(self, createCombiner, mergeValue, mergeCombiners,
                       numPartitions=None, partitionFunc=portable_hash):
          """
          Generic function to combine the elements for each key using a custom
          set of aggregation functions.
  
          Turns an RDD[(K, V)] into a result of type RDD[(K, C)], for a "combined
          type" C.
  
          Users provide three functions:
  
              - C{createCombiner}, which turns a V into a C (e.g., creates
                a one-element list)
              - C{mergeValue}, to merge a V into a C (e.g., adds it to the end of
                a list)
              - C{mergeCombiners}, to combine two C's into a single one (e.g., merges
                the lists)
  
          To avoid memory allocation, both mergeValue and mergeCombiners are allowed to
          modify and return their first argument instead of creating a new C.
  
          In addition, users can control the partitioning of the output RDD.
  
          .. note:: V and C can be different -- for example, one might group an RDD of type
              (Int, Int) into an RDD of type (Int, List[Int]).
  
          >>> x = sc.parallelize([("a", 1), ("b", 1), ("a", 2)])
          >>> def to_list(a):
          ...     return [a]
          ...
          >>> def append(a, b):
          ...     a.append(b)
          ...     return a
          ...
          >>> def extend(a, b):
          ...     a.extend(b)
          ...     return a
          ...
          >>> sorted(x.combineByKey(to_list, append, extend).collect())
          [('a', [1, 2]), ('b', [1])]
          """
          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)