Code sketched for rmr vector API

rmr-vec-api.R

#predicate, group and aggregate are user defined functions
#it is assumed a vectorized version is used when needed

#pass through
mapreduce(input,
          map = function(k,v) keyval(k,v))
#vec version
#nrecs says how many records to process in one map
mapreduce(input,
          map = function(k,v) vec.keyval(k,v), 
          nrecs = 1000)

#filter
mapreduce(input,
          map = function(k,v) if(predicate(k,v)) keyval(k,v))
#vec version
mapreduce(input, 
          map = function(k,v) {filter = predicate(k,v); 
                               vec.keyval(k[filter], v[filter])},
          nrecs = 1000)
#vec version, structured case
mapreduce(input, 
          map = function(k,v) {filter = predicate(k,v); 
                               vec.keyval(k[filter,], v[filter,])},
          nrecs = 1000,
          map.on.data.frame = T)

#select TODO

#bigsum
mapreduce(input, 
          map  = function(k,v) keyval(group(k,v),v), 
          reduce = function(k, vv) keyval(k, sum(unlist(vv))))
#vec version
mapreduce(input, 
          map  = function(k,v) vec.keyval(group(k,v),v), 
          reduce = function(k, vv) keyval(k, sum(unlist(vv))), 
          nrecs = 1000)
#vec version, structured case
mapreduce(input, 
          map  = function(k,v) vec.keyval(group(k,v),v), 
          reduce = function(k, vv) keyval(k, sum(vv)) , 
          map.on.data.frame = TRUE
          reduce.on.data.frame = TRUE,
          nrecs = 1000)

#embarrassingly parallel
mapreduce(input, 
          map = function(k,v) keyval(group(k,v), v),
          reduce = function(k, vv) keyval(k, aggregate(vv)))
#vec version
mapreduce(input, 
          map = function(k,v) vec.keyval(group(k,v), v),
          reduce = function(k, vv) vec.keyval(k, aggregate(vv)),
          nrecs = 1000)
#vec version, structured case
mapreduce(input, 
          map = function(k,v) vec.keyval(group(k,v), v),
          reduce = function(k, vv) vec.keyval(k, some.function(vv))
          map.on.data.frame = TRUE,
          reduce.on.data.frame = TRUE,
          nrecs = 1000)