def sort_by(key_function=nil, ascending=true, num_partitions=nil)
      key_function   ||= 'lambda{|x| x}'
      num_partitions ||= default_reduce_partitions

      command_klass = Spark::Command::SortByKey

      # Allow spill data to disk due to memory limit
      # spilling = config['spark.shuffle.spill'] || false
      spilling = false
      memory = ''

      # Set spilling to false if worker has unlimited memory
      if memory.empty?
        spilling = false
        memory   = nil
      else
        memory = to_memory_size(memory)
      end

      # Sorting should do one worker
      if num_partitions == 1
        rdd = self
        rdd = rdd.coalesce(1) if partitions_size > 1
        return rdd.new_rdd_from_command(command_klass, key_function, ascending, spilling, memory, serializer)
      end

      # Compute boundary of collection
      # Collection should be evenly distributed
      # 20.0 is from scala RangePartitioner (for roughly balanced output partitions)
      count = self.count
      sample_size = num_partitions * 20.0
      fraction = [sample_size / [count, 1].max, 1.0].min
      samples = self.sample(false, fraction, 1).map(key_function).collect
      samples.sort!
      # Reverse is much faster than reverse sort_by
      samples.reverse! if !ascending

      # Determine part bounds
      bounds = determine_bounds(samples, num_partitions)

      shuffled = _partition_by(num_partitions, Spark::Command::PartitionBy::Sorting, key_function, bounds, ascending, num_partitions)
      shuffled.new_rdd_from_command(command_klass, key_function, ascending, spilling, memory, serializer)
    end
def sort_by(key_function=nil, ascending=true, num_partitions=nil)
      key_function   ||= 'lambda{|x| x}'
      num_partitions ||= default_reduce_partitions

      command_klass = Spark::Command::SortByKey

      # Allow spill data to disk due to memory limit
      # spilling = config['spark.shuffle.spill'] || false
      spilling = false
      memory = ''

      # Set spilling to false if worker has unlimited memory
      if memory.empty?
        spilling = false
        memory   = nil
      else
        memory = to_memory_size(memory)
      end

      # Sorting should do one worker
      if num_partitions == 1
        rdd = self
        rdd = rdd.coalesce(1) if partitions_size > 1
        return rdd.new_rdd_from_command(command_klass, key_function, ascending, spilling, memory, serializer)
      end

      # Compute boundary of collection
      # Collection should be evenly distributed
      # 20.0 is from scala RangePartitioner (for roughly balanced output partitions)
      count = self.count
      sample_size = num_partitions * 20.0
      fraction = [sample_size / [count, 1].max, 1.0].min
      samples = self.sample(false, fraction, 1).map(key_function).collect
      samples.sort!
      # Reverse is much faster than reverse sort_by
      samples.reverse! if !ascending

      # Determine part bounds
      bounds = determine_bounds(samples, num_partitions)

      shuffled = _partition_by(num_partitions, Spark::Command::PartitionBy::Sorting, key_function, bounds, ascending, num_partitions)
      shuffled.new_rdd_from_command(command_klass, key_function, ascending, spilling, memory, serializer)
    end
Sorts this RDD by the given key_function

 This is a different implementation than spark. Sort by doesn't use
 key_by method first. It can be slower but take less memory and
 you can always use map.sort_by_key

 == Example:
   rdd = $sc.parallelize(["aaaaaaa", "cc", "b", "eeee", "ddd"])

   rdd.sort_by.collect
   # => ["aaaaaaa", "b", "cc", "ddd", "eeee"]

   rdd.sort_by(lambda{|x| x.size}).collect
   # => ["b", "cc", "ddd", "eeee", "aaaaaaa"]