def take(count)
      buffer = []

      parts_count = self.partitions_size
      # No parts was scanned, yet
      last_scanned = -1

      while buffer.empty?
        last_scanned += 1
        buffer += context.run_job_with_command(self, [last_scanned], true, Spark::Command::Take, 0, -1)
      end

      # Assumption. Depend on batch_size and how Spark divided data.
      items_per_part = buffer.size
      left = count - buffer.size

      while left > 0 && last_scanned < parts_count
        parts_to_take = (left.to_f/items_per_part).ceil
        parts_for_scanned = Array.new(parts_to_take) do
          last_scanned += 1
        end

        # We cannot take exact number of items because workers are isolated from each other.
        # => once you take e.g. 50% from last part and left is still > 0 then its very
        # difficult merge new items
        items = context.run_job_with_command(self, parts_for_scanned, true, Spark::Command::Take, left, last_scanned)
        buffer += items

        left = count - buffer.size
        # Average size of all parts
        items_per_part = [items_per_part, items.size].reduce(0){|sum, x| sum + x.to_f/2}
      end

      buffer.slice!(0, count)
    end
def take(count)
      buffer = []

      parts_count = self.partitions_size
      # No parts was scanned, yet
      last_scanned = -1

      while buffer.empty?
        last_scanned += 1
        buffer += context.run_job_with_command(self, [last_scanned], true, Spark::Command::Take, 0, -1)
      end

      # Assumption. Depend on batch_size and how Spark divided data.
      items_per_part = buffer.size
      left = count - buffer.size

      while left > 0 && last_scanned < parts_count
        parts_to_take = (left.to_f/items_per_part).ceil
        parts_for_scanned = Array.new(parts_to_take) do
          last_scanned += 1
        end

        # We cannot take exact number of items because workers are isolated from each other.
        # => once you take e.g. 50% from last part and left is still > 0 then its very
        # difficult merge new items
        items = context.run_job_with_command(self, parts_for_scanned, true, Spark::Command::Take, left, last_scanned)
        buffer += items

        left = count - buffer.size
        # Average size of all parts
        items_per_part = [items_per_part, items.size].reduce(0){|sum, x| sum + x.to_f/2}
      end

      buffer.slice!(0, count)
    end
Take the first num elements of the RDD.

 It works by first scanning one partition, and use the results from
 that partition to estimate the number of additional partitions needed
 to satisfy the limit.

 == Example:
   rdd = $sc.parallelize(0..100, 20)
   rdd.take(5)
   # => [0, 1, 2, 3, 4]