<commit_before>require "sourcify"

# Resilient Distributed Dataset

module Spark
  class RDD

    attr_reader :jrdd, :context, :serializer

    def initialize(jrdd, context, serializer)
      @jrdd = jrdd
      @context = context
      @serializer = serializer

      @cached = false
      @checkpointed = false
    end



    # =======================================================================    
    # Variables 
    # =======================================================================   

    def default_reduce_partitions
      if @context.conf.contains("spark.default.parallelism")
        @context.default_parallelism
      else
        @jrdd.partitions.size
      end
    end

    def id
      @jrdd.id
    end

    def cached?
      @cached
    end

    def checkpointed?
      @checkpointed
    end



    # =======================================================================
    # Computing functions
    # =======================================================================     

    #
    # Return an array that contains all of the elements in this RDD.
    #
    def collect
      # @serializer.load(jrdd.collect.iterator)
      @serializer.load(jrdd.collect.to_a)
    end


    #
    # Return a new RDD by applying a function to all elements of this RDD.
    #
    # rdd = $sc.parallelize(0..5)
    # rdd.map(lambda {|x| x*2}).collect
    # => [0, 2, 4, 6, 8, 10]
    #
    def map(f)
      function = [to_source(f), "Proc.new {|iterator| iterator.map{|i| @__function__.call(i)} }"]
      PipelinedRDD.new(self, function)
    end

    #
    # Return a new RDD by first applying a function to all elements of this
    # RDD, and then flattening the results.
    #
    # rdd = $sc.parallelize(0..5)
    # rdd.flat_map(lambda {|x| [x, 1]}).collect
    # => [0, 1, 2, 1, 4, 1, 6, 1, 8, 1, 10, 1]
    #
    def flat_map(f)
      function = [to_source(f), "Proc.new {|iterator| iterator.flat_map{|i| @__function__.call(i)} }"]
      PipelinedRDD.new(self, function)
    end

    #
    # Return a new RDD by applying a function to each partition of this RDD.
    #
    # rdd = $sc.parallelize(0..10, 2)
    # rdd.map_partitions(lambda{|part| part.reduce(:+)}).collect
    # => [15, 40]
    #
    def map_partitions(f)
      function = [to_source(f), "Proc.new {|iterator| @__function__.call(iterator) }"]
      PipelinedRDD.new(self, function)
    end

    #
    # Return a new RDD by applying a function to each partition of this RDD, while tracking the index
    # of the original partition.
    #
    # rdd = $sc.parallelize(0...4, 4)
    # rdd.map_partitions_with_index(lambda{|part, index| part[0] * index}).collect
    # => [0, 1, 4, 9]
    #
    def map_partitions_with_index(f)
      function = [to_source(f), "Proc.new {|iterator, index| @__function__.call(iterator, index) }"]
      PipelinedRDD.new(self, function)
    end


    # def reduce_by_key(f, num_partitions=nil)
    #   combine_by_key(lambda {|x| x}, f, f, num_partitions)
    # end

    # def combine_by_key(create_combiner, merge_value, merge_combiners, num_partitions=nil)
    #   num_partitions ||= default_reduce_partitions
    # end



    # Aliases
    alias_method :flatMap, :flat_map
    alias_method :mapPartitions, :map_partitions<commit_msg>RDD: +filter<commit_after>require "sourcify"

# Resilient Distributed Dataset

module Spark
  class RDD

    attr_reader :jrdd, :context, :serializer

    def initialize(jrdd, context, serializer)
      @jrdd = jrdd
      @context = context
      @serializer = serializer

      @cached = false
      @checkpointed = false
    end



    # =======================================================================    
    # Variables 
    # =======================================================================   

    def default_reduce_partitions
      if @context.conf.contains("spark.default.parallelism")
        @context.default_parallelism
      else
        @jrdd.partitions.size
      end
    end

    def id
      @jrdd.id
    end

    def cached?
      @cached
    end

    def checkpointed?
      @checkpointed
    end



    # =======================================================================
    # Computing functions
    # =======================================================================     

    #
    # Return an array that contains all of the elements in this RDD.
    #
    def collect
      # @serializer.load(jrdd.collect.iterator)
      @serializer.load(jrdd.collect.to_a)
    end


    #
    # Return a new RDD by applying a function to all elements of this RDD.
    #
    # rdd = $sc.parallelize(0..5)
    # rdd.map(lambda {|x| x*2}).collect
    # => [0, 2, 4, 6, 8, 10]
    #
    def map(f)
      function = [to_source(f), "Proc.new {|iterator| iterator.map{|i| @__function__.call(i)} }"]
      PipelinedRDD.new(self, function)
    end

    #
    # Return a new RDD by first applying a function to all elements of this
    # RDD, and then flattening the results.
    #
    # rdd = $sc.parallelize(0..5)
    # rdd.flat_map(lambda {|x| [x, 1]}).collect
    # => [0, 1, 2, 1, 4, 1, 6, 1, 8, 1, 10, 1]
    #
    def flat_map(f)
      function = [to_source(f), "Proc.new {|iterator| iterator.flat_map{|i| @__function__.call(i)} }"]
      PipelinedRDD.new(self, function)
    end

    #
    # Return a new RDD by applying a function to each partition of this RDD.
    #
    # rdd = $sc.parallelize(0..10, 2)
    # rdd.map_partitions(lambda{|part| part.reduce(:+)}).collect
    # => [15, 40]
    #
    def map_partitions(f)
      function = [to_source(f), "Proc.new {|iterator| @__function__.call(iterator) }"]
      PipelinedRDD.new(self, function)
    end

    #
    # Return a new RDD by applying a function to each partition of this RDD, while tracking the index
    # of the original partition.
    #
    # rdd = $sc.parallelize(0...4, 4)
    # rdd.map_partitions_with_index(lambda{|part, index| part[0] * index}).collect
    # => [0, 1, 4, 9]
    #
    def map_partitions_with_index(f)
      function = [to_source(f), "Proc.new {|iterator, index| @__function__.call(iterator, index) }"]
      PipelinedRDD.new(self, function)
    end

    #
    # Return a new RDD containing only the elements that satisfy a predicate.
    #
    # rdd = $sc.parallelize(0..10)
    # rdd.filter(lambda{|x| x.even?}).collect
    # => [0, 2, 4, 6, 8, 10]
    #
    def filter(f)
      function = [to_source(f), "Proc.new {|iterator| iterator.select{|i| @__function__.call(i)} }"]
      PipelinedRDD.new(self, function)
    end


    # def reduce_by_key(f, num_partitions=nil)
    #   combine_by_key(lambda {|x| x}, f, f, num_partitions)
    # end

    # def combine_by_key(create_combiner, merge_value, merge_combiners, num_partitions=nil)
    #   num_partitions ||= default_reduce_partitions
    # end



    # Aliases
    alias_method :flatMap, :flat_map
    alias_method :mapPartitions, :map_partitions
{'fraction_non_alphanumeric': 0.19522839685772475, 'fraction_numerical': 0.013674716322374164, 'mean_word_length': 2.8152053274139845, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 12, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7172984', 'n_tokens_mistral': 2413, 'n_tokens_neox': 2222, 'n_words': 543}