Contents

module Enumerable    
  def mapreduce(klass=nil,options={},&block)               
    data = []
    if self.is_a?(Hash)
      self.each {|k,v| data << {k => v}}
    else
      data = self
    end
    jobopts = {
      :mappers                => 20000,
      :map_data               => data,
      :name                   => "#{klass} Enumerable MASTER",
      :map_name               => "#{klass} Enumerable MAP",
      :reduce_name            => "#{klass} Enumerable REDUCE",
      :map_timeout            => 3600,
      :reduce_timeout         => 3600,
      :master_timeout         => 3600,
      :master_result_timeout  => 3600
    }                                

    jobopts[:map_reduce_class] = klass.to_s if klass

    options.each { |k,v| jobopts[k] = v }
    if block_given?
      jobopts[:map] = block
    end                               
    
    if block_given? or not jobopts[:async]
      job = Skynet::Job.new(jobopts.merge(:local_master => true))
    else
      job = Skynet::AsyncJob.new(jobopts)
    end
    job.run    
  end  
end

Version data entries

1 entries across 1 versions & 1 rubygems

Version	Path
skynet-0.9.2	lib/skynet/skynet_ruby_extensions.rb