nodes: 0, module; 1, function_definition; 2, function_name:_custom_rdd_reduce; 3, parameters; 4, block; 5, identifier:self; 6, identifier:reduce_func; 7, expression_statement; 8, function_definition; 9, expression_statement; 10, return_statement; 11, comment:"""Provides a custom RDD reduce which preserves ordering if the RDD has
          been sorted. This is useful for us because we need this functionality
          as many pandas operations support sorting the results. The standard
          reduce in PySpark does not have this property.  Note that when PySpark
          no longer does partition reduces locally this code will also need to
          be updated."""; 12, function_name:accumulating_iter; 13, parameters; 14, block; 15, assignment; 16, call; 17, identifier:iterator; 18, expression_statement; 19, for_statement; 20, if_statement; 21, identifier:vals; 22, call; 23, identifier:reduce; 24, argument_list; 25, assignment; 26, identifier:obj; 27, identifier:iterator; 28, block; 29, comparison_operator:acc is not None; 30, block; 31, attribute; 32, argument_list; 33, identifier:accumulating_iter; 34, identifier:vals; 35, identifier:acc; 36, None; 37, if_statement; 38, identifier:acc; 39, None; 40, expression_statement; 41, call; 42, identifier:collect; 43, comparison_operator:acc is None; 44, block; 45, else_clause; 46, yield; 47, attribute; 48, argument_list; 49, identifier:acc; 50, None; 51, expression_statement; 52, block; 53, identifier:acc; 54, attribute; 55, identifier:mapPartitions; 56, identifier:accumulating_iter; 57, assignment; 58, expression_statement; 59, identifier:self; 60, identifier:_rdd; 61, identifier:acc; 62, identifier:obj; 63, assignment; 64, identifier:acc; 65, call; 66, identifier:reduce_func; 67, argument_list; 68, identifier:acc; 69, identifier:obj
code: def _custom_rdd_reduce(self, reduce_func):
          """Provides a custom RDD reduce which preserves ordering if the RDD has
          been sorted. This is useful for us because we need this functionality
          as many pandas operations support sorting the results. The standard
          reduce in PySpark does not have this property.  Note that when PySpark
          no longer does partition reduces locally this code will also need to
          be updated."""
          def accumulating_iter(iterator):
              acc = None
              for obj in iterator:
                  if acc is None:
                      acc = obj
                  else:
                      acc = reduce_func(acc, obj)
              if acc is not None:
                  yield acc
          vals = self._rdd.mapPartitions(accumulating_iter).collect()
          return reduce(accumulating_iter, vals)