Passing Global variable in map expression

[Shekhar]

Hi,
I am trying to pass som global variables inside the map expression for
initializing the centers in k-means algorithm.
My intention is to do something like this.

Script#1:

c1<-c(1,2)
c2<-c(2,1)
c3<-c(3,1)
#map-reduce-job-script.R will use the variable c1, c2, c3
source("map-reduce-job-script.R")

This is not working, it throws me error in readbin

I have tried something like this:

1. Written the centers in a file and stored locally

1 2
2 1
3 1

2. Then inside the map function i tried reading the file but again
this is giving me the error in readbin

Then i stored the file on hdfs but then i got the error in the funtion
itself and i cannot use rhread to read the file because its a text
file and rhread requires sequence file.

well i am thinking to write a map reduce job for converting the text
file to sequence file and in the original script i will try reading
the file on the hadoop.

I have a feeling that reading another file from the hdfs has to do
with R session also, but i am not sure.
But if it works then i guess we can write iterative map reduce job, by
calling the script iteratively

[Saptarshi Guha]

There are several ways.

1. For a large number of variables (or large in size e.g. a few MB)

rhsave(n1,n2,n3, file="/path-on-hdfs/data.Rdata")

map.setup = expression({
load("data.Rdata") # no need to give full path
})

rhmr(...,setup=list(map=map.setup),
shared.files=c("/path-on-hdfs/data.Rdata"),....

2. Serialize as strings and pass via mapred

mapred=list()
mapred$c1 = rawToChar(serialize(c1,NULL,ascii=TRUE))
mapred$c2 = rawToChar(serialize(c2,NULL,ascii=TRUE))

map.setup = expression({
c1 = unserialize( charToRaw( Sys.getenv("c1") ))
c2 = unserialize( charToRaw( Sys.getenv("c2") ))
})

rhmr(,.., mapred=mapred,setup=list(map=map.setup))

3. Modify R code using bquote (you should be comfortable with
modifying code at run time)

map = as.expression(bquote({
c1 <- .(c1)
c2 <- .(c2)
.....}, list(c1=c1,c2=c2)))

rhmr(map=map,...)

Note, this code is parsed and modified before being sent to Hadoop. c1
and c2 are hardcoded into the source. Try this

c1=matrix(runif(10),ncol=2)
c2=matrix(runif(10),ncol=2)
map = as.expression(bquote({
c1 <- .(c1)
c2 <- .(c2)
}, list(c1=c1,c2=c2)))

and type map, you'll see c1 and c2 in the source. Use this for i)
small number of global variables ii) small sized variables (< MB)

Regards

Saptarshi

[Shekhar]

Hi,
Thank you very much for the useful suggestions.

Actually i want to do something like this:

i have two scripts : #1 and #2

Script #1:

library(Rhipe)

# providing the initial centres
x<-c(1,2)
y<-c(3,4)
z<-c(5,6)

centerList<-list(x,y,z)

rhsave(centerList,file="data/centers.RData")

source("Script2.R")
# after the successful job location read the output of script 2 and
calculate the new centers,
#and replace the old centers with the new centers and delete the
output of the script 2


Now Script 2:
I will be loading the data and execute the map reduce job.
library(Rhipe)
rhinit()
aton<-new.env();
map.setup = expression({
load("centers.Rdata")
center<-function(N)
{
aton[[N]]
}
})

map<-expression({
y<-do.call("rbind",lapply(map.values,function(r){
as.numeric(strsplit(r," ")[[1]])
}))

if(nrow(y) > 0)
{
col.x = y[,1]
col.y = y[,2]
c1=center(1)
c2<-center(2)
c3<-center(c)
centerMat<-rbind(c1,rbind(c2,c3))
#forming the full data frame
d<-data.frame(col.x=col.x,col.y=col.y,stringsAsFactors=FALSE)
#Appeding the center matrix to the top of the data frame
dmat<-rbind(centerMat,as.matrix(d))
#Finding the euclidean distance
reqMat<-(as.matrix(dist(dmat,method="euclidean")))[4:nrow(y),1:3]
#creating three data frame for three different centers
d1<-data.frame()#data frame for centre1
d2<-data.frame()#data frame for centre2
d3<-data.frame()#data frame for centre3

for( i in 1:nrow(reqMat) )
{
minimum = which.min(reqMat[i,])

if(minimum==1) d1<-rbind(d1,d[i, ])
else if(minimum==2) d2<-rbind(d2,d[i, ])
else if(minimum==3) d3<-rbind(d3,d[i, ])
}
rhcollect(c1,d1)
rhcollect(c2,d2)
rhcollect(c3,d3)
}
})

reduce<-expression(
pre = { collect<-NULL } ,
reduce = {
collect<-
rbind(collect,do.call("rbind",reduce.values))
},
post = {
rhcollect(reduce.key,collect)
}

)



mapred<-
list(rhipe_map_buff_size=20,mapred.task.timeout=0,mapred.reduce.tasks=1,mapred.job.tracker='local')

ifolder="/km"

ofolder="/out01"

z<-
rhmr(map=map,reduce=reduce,inout=c("text","sequence"),ifolder=ifolder,
ofolder=ofolder,setup=list(map=map.setup),
shared=c("data/centers.Rdata"),mapred=mapred,jobname="kairo")
rhex(z)


Am i doing something wrong here??? please correct me

Why always one type of error is thrown : Error in readBin ?
Well how to debug this.. i dont get a clue why this error is thrown..
11

[Shekhar]

Inline with the above equation is it possible to pass the varibles
from one script to another?

> list(rhipe_map_buff_size=20,mapred.task.timeout=0,mapred.reduce.tasks=1,map­red.job.tracker='local')

[Shekhar]

Hi,
I figured out by writing so many map reduce jobs that error in readbin
is thrown for the following
reasons:
1. Input and output folder are wrongly mentioned
For example: if the original input file created is "input/file"
and in the code you have provided as ifolder="/input/file".

In this case readBin error is thrown.

2. If you are storing the data on HDFS using the function rhsave like
rhsave(x,y,file="data/dat.Rdata")
and in the map set up expression you are calling as load("dat.RData")
{Note the diff Rdata and RData "D" is caps), in this case error is
thrown

If you do vice versa also i.e. storing the file on HDFS as
rhsave(x,"data/dat.RData") and accessing in the map set up expression
as load("dat.Rdata") then also error is thrown

3. If you are trying to access the matrix or data frame with incorrect
index then also error is thrown

Well in most of the cases this error is thrown if there is some
logical problem in the code only.
I guess there are other cases aswell where this error is thrown, i
think you can update me more on this kind of error.

Well cant these error be made more specific to the problem?

Regards,
Shekhar

[Saptarshi Guha]

I've said this before.

1. If you use rhex(z) and call the script using source, you wont see
errors. The error code is because there
were hadoop/R errors. The errors are printed out by hadoop and using
source will hide these. If you use rhex(z)
then either
a) run rhex(z) interactively at the prompt
b) if you call it via source, then go to the jobtracker and look for the errors

2. You can also try this (either using source or interactively)

j <- rhex(z, async=TRUE)
jr <- rhstatus(j, mon.sec=5,autokill=TRUE)

the last line monitors the job and will kill it if there are R errors
(but will /not/ kill it if there are Hadoop errors, since Hadoop
errors need not be fatal)
Also, it has attributes that contain the R / Hadoop errors.

[Jeremiah Rounds]

> 2. Serialize as strings and pass via mapred
>
> mapred=list()
> mapred$c1 = rawToChar(serialize(c1,NULL,ascii=TRUE))
> mapred$c2 = rawToChar(serialize(c2,NULL,ascii=TRUE))
>
> map.setup = expression({
>     c1 = unserialize( charToRaw( Sys.getenv("c1") ))
>     c2 = unserialize( charToRaw( Sys.getenv("c2") ))
>
> })
>
> rhmr(,.., mapred=mapred,setup=list(map=map.setup))
>

Out of curiosity can you think of a way to automate this so I don't
have to do it in a custom way each time. I am thinking of something
like what follows, but I have not gotten this to work yet. I think it
is because I am not entirely clear on the evaluation rules of R.


The idea is list.setup holds a list of what ever you want setup and
list.setup is seen from inside the map reduce operations. Instead of
thinking about expressions, one thinks about a list of things they
want inside of map/reduce.

rhmrWithListSetup = function(list.setup, ...){
mc = match.call()
mc = as.list(mc)
if(!is.null(mc$setup)){
cat("Error: list setup currently incompatable with setup argument.
\n")
}

mc[[1]] = NULL # removes call from list
mc$list.setup = NULL #removes from list
list.setup = rawToChar(serialize(list.setup,NULL,ascii=TRUE))
print(list.setup)
if(!is.null(mc$mapred)){
mc$mapred$list.setup = list.setup
}else{
mc$mapred = list()
mc$mapred$list.setup = list.setup
}
setup = expression({
list.setup =
unserialize( charToRaw( Sys.getenv("list.setup") ))
rhstatus(list.setup)
})
mc$setup = setup
return(do.call("rhmr", mc))


}

But I haven't got this working yet.

[Saptarshi Guha]

what isn't working? also, you dont' want rhstatus in mapred jobs,
(the function is not available)

[Saptarshi Guha]

scratch the last bit. I was confused. rhstatus is available in mapreduce jobs.

[Jeremiah Rounds]

rhstatus shouldn't have been in there. I was tinkering with trying to
print some stuff out to get a picture of what went wrong. This was
part of a larger effort. I will see if I had can distill it down.

[Jeremiah Rounds]

Hmm distilling it down to a minimum test case revealed that it
worked. I changed it to pass environments instead of list and
attached that environment in the setup prior to map/reduce. Seems to
work. Gives an output of 5.0 for each time the mapper was invoked.


rhmrWithEnvSetup = function(env.setup, ...){

mc = match.call()
mc = as.list(mc)
if(!is.null(mc$setup)){

cat("Error: env setup currently incompatable with setup argument.

\n")
}

mc[[1]] = NULL # removes call from list

mc$env.setup = NULL #removes from list
env.setup = rawToChar(serialize(env.setup,NULL,ascii=TRUE))
print(env.setup)
if(!is.null(mc$mapred)){
mc$mapred$env.setup = env.setup
}else{
mc$mapred = list()
mc$mapred$env.setup = env.setup
}
setup = expression({
env.setup = unserialize( charToRaw( Sys.getenv("env.setup") ))
attach(env.setup)
print(ls())

})
mc$setup = setup
return(do.call("rhmr", mc))


}

mapper = expression({
rhcollect(1,test.function())
})


env.setup = new.env()
env.setup$test.function = function(){
return(5)
}
DIR = getwd()
args = list()
args$map = mapper
args$env.setup = env.setup
args$ifolder = paste(DIR,"input",sep="/")
args$ofolder = paste(DIR,"output", sep="/")
args$inout = c("text","text")
args$mapred = list(mapred.job.tracker='local')
mr = do.call("rhmrWithEnvSetup", args)
rhex(mr)