RevolutionAnalytics/RHadoop — Issue #21: Hadoop streaming issue (apparently) with mapreduce
Author: kagharpure
Created: 2011-10-20 18:34:52+00:00
State: open

Hi all,

Firstly, I'd like to apologize for not sticking to gfm conventions: it's almost midnight, my brain is nowhere near the functional level required to figure something new out and I'm posting here for the first time. Now:

I am somewhat familiar with R (~1.5 years), fairly new to Hadoop and very new to RHadoop. I've recently been trying to run a couple of simple linear regression codes written using RHadoop; however, I seem to be running repeatedly into what seems to be a hadoop streaming issue. Before I jump into it, I shall explain the codes and where I'm running them a little.

Both of the codes mentioned above expect an input csv on the HDFS with no headers and only containing values for the predictor variables and the dependent variable, which must be the last variable (counting left to right)

Since I'm not entirely sure yet of the way RHadoop works, I wrote both codes with different assumptions on RHadoop's working. One code is based on the assumption that text files will be processed line-by-line by the mapper and the second one is a logical rip-off from an equivalent Rhipe code, where the assumption is that for each mapper there will exist (magically) a list of values (a la map.values in Rhipe).

First code (assuming mapper processes line-by-line):

### Script: RHadoopReg.R ###
# load required libraries
library(rhdfs)
library(rmr)

# define function rHdp.lm
rHdp.lm <- function(input,output = NULL){

# define mapper as function
map <- function(k,v){
# split input line on commas
val <- strsplit(v, ',')[[1]]

# extract predictors and bind
# 1 to the right
x <- c(1,as.numeric(val[-length(val)]))

# extract dependent
y <- as.numeric(val[length(val)])

# compute outer of x and x
# for m predictors, this should
# yield a (m + 1)x(m + 1) matrix
val1 <- outer(x,x)

# multiply x with dependent
# this should yield a vector
# of length (m + 1)
val2 <- x*y

# cbind val1 and val2 and write
# out with NULL key
keyval('dummy_key',cbind(val1,val2))
}

# define reducer as function
reduce <- function(k,vv){

# initialize val as zero
val <- 0

# loop through each (matrix)
# element of vv and take matrix
# sums
for(i in 1:length(vv)){
val = val + as.numeric(vv[[i]])
}

# extract t(x)*x
x <- val[,-dim(val)[2]]

# extract t(x)*y
y <- val[,dim(val)[2]]

# use solve to get betas
betas <- solve(as.matrix(x),as.matrix(y))

# write out with key betas
keyval('betas',betas)
}

# define as a map-reduce job
mapreduce(input = input,
output = output,
map = map,
reduce = reduce)
}

rHdp.lm('/user/hadoop/regData','/user/hadoop/rTestRegOut')

Second code (assuming mappers process lists of values):

### Script: RHadoopReg1.R ###
# load required libraries
library(rhdfs)
library(rmr)

# define function rHdp.lm1
rHdp.lm1 <- function(input,output = NULL){

# define mapper as function
map <- function(k,v){
# split input line on commas
datMat <- do.call('rbind',lapply(v, function(r){
as.numeric(strsplit(r,',')[[1]])
}))

# extract dependent
y <- as.matrix(datMat[,dim(datMat)[2]])

# extract predictors and bind
# 1 to the right
x <- cbind(rep(1,dim(datMat)[1]),as.matrix(datMat[,-dim(datMat)[2]]))

# compute crossprod of x and x
# for m predictors, this should
# yield a (m + 1)x(m + 1) matrix
val1 <- crossprod(as.matrix(x),as.matrix(x))

# multiply x with dependent
# this should yield a vector
# of length (m + 1)
val2 <- crossprod(as.matrix(x),as.matrix(y))

# cbind val1 and val2 and write
# out with NULL key
keyval('temp_out',cbind(val1,val2))
}

# define reducer as function
reduce <- function(k,vv){

# initialize val as zero
val <- 0

# loop through each (matrix)
# element of vv and take matrix
# sums
for(i in 1:length(vv)){
val = val + as.numeric(vv[[i]])
}

# extract t(x)*x
x <- val[,-dim(val)[2]]

# extract t(x)*y
y <- val[,dim(val)[2]]

# use solve to get betas
betas <- solve(as.matrix(x),as.matrix(y))

# write out with key betas
keyval('betas',betas)
}

# define as a map-reduce job
mapreduce(input = input,
output = output,
map = map,
reduce = reduce)
}

rHdp.lm1('/user/hadoop/regData/','/user/hadoop/rTestRegOut1')

I've tried running these codes both on a 14-node hadoop cluster, with nodes running hadoop-0.20.2.append on either Ubuntu Lucid or Maverick, and a single node installation, running on Ubuntu Lucid which itself runs on VMPlayer on Windows 7.

Now, upon sourcing either of these two codes in R, I repeatedly keep running into the following error:

> source("RHadoopReg.R")
packageJobJar: [/tmp/RtmpIKd3Fr/rhstr.map287d3049, /tmp/RtmpIKd3Fr/rhstr.reduce20ae0aad, /tmp/RtmpIKd3Fr/rmrParentEnv, /tmp/RtmpIKd3Fr/rmrLocalEnv, /usr/local/hadoop/hadoop-datastore/hadoop-hadoop/hadoop-unjar7948228214583466863/] [] /tmp/streamjob2980539178006618836.jar tmpDir=null
11/10/20 10:51:33 INFO mapred.FileInputFormat: Total input paths to process : 1
11/10/20 10:51:34 INFO streaming.StreamJob: getLocalDirs(): [/usr/local/hadoop/hadoop-datastore/hadoop-hadoop/mapred/local]
11/10/20 10:51:34 INFO streaming.StreamJob: Running job: job_201110200316_0004
11/10/20 10:51:34 INFO streaming.StreamJob: To kill this job, run:
11/10/20 10:51:34 INFO streaming.StreamJob: /usr/local/hadoop/hadoop/bin/../bin/hadoop job -Dmapred.job.tracker=localhost:54311 -kill job_201110200316_0004
11/10/20 10:51:35 INFO streaming.StreamJob: Tracking URL: http://localhost.localdomain:50030/jobdetails.jsp?jobid=job_201110200316_0004
11/10/20 10:51:36 INFO streaming.StreamJob: map 0% reduce 0%
11/10/20 10:52:38 INFO streaming.StreamJob: map 100% reduce 0%
11/10/20 10:52:46 INFO streaming.StreamJob: map 0% reduce 0%
11/10/20 10:53:17 INFO streaming.StreamJob: map 100% reduce 100%
11/10/20 10:53:17 INFO streaming.StreamJob: To kill this job, run:
11/10/20 10:53:17 INFO streaming.StreamJob: /usr/local/hadoop/hadoop/bin/../bin/hadoop job -Dmapred.job.tracker=localhost:54311 -kill job_201110200316_0004
11/10/20 10:53:17 INFO streaming.StreamJob: Tracking URL: http://localhost.localdomain:50030/jobdetails.jsp?jobid=job_201110200316_0004
11/10/20 10:53:17 ERROR streaming.StreamJob: Job not Successful!
11/10/20 10:53:17 INFO streaming.StreamJob: killJob...
Streaming Job Failed!
Error in rhstream(map = map, reduce = reduce, reduceondataframe = reduceondataframe, :
hadoop streaming failed with error code 1

Further exploring the trace on the jobtracker url, this is what I find (for each attempt):

java.lang.RuntimeException: PipeMapRed.waitOutputThreads(): subprocess failed with code 1 at org.apache.hadoop.streaming.PipeMapRed.waitOutputThreads(PipeMapRed.java:311) at org.apache.hadoop.streaming.PipeMapRed.mapRedFinished(PipeMapRed.java:545) at org.apache.hadoop.streaming.PipeMapper.close(PipeMapper.java:132) at org.apache.hadoop.mapred.MapRunner.run(MapRunner.java:57) at org.apache.hadoop.streaming.PipeMapRunner.run(PipeMapRunner.java:36) at org.apache.hadoop.mapred.MapTask.runOldMapper(MapTask.java:358) at org.apache.hadoop.mapred.MapTask.run(MapTask.java:307) at org.apache.hadoop.mapred.Child.main(Child.java:170)

While this trace seems somewhat common on Google, I cannot seem to find anything specifically about it in an RHadoop context.

Additionally, I've also been observing some strange behavior from rhdfs.

This set of commands run fine:

> to.dfs(lapply(1:100, function(i){eps = rnorm(1,sd=10);keyval(i, list(x=c(i,i+eps),y=2*(eps > 0) - 1))}),"/user/hadoop/logreg")
> a <- from.dfs("/user/hadoop/logreg")

However, when I try the same trick with a file already on the HDFS (put there previously using hadoop fs -copyFromLocal), I seem to run into a strange error:

> b <- from.dfs("/user/hadoop/regData")
Error in file(con, "r") : cannot open the connection

I initially suspected that this might have something to do with permissions and I was partially right; when I tried the first set of commands as root, I got the following error:

> to.dfs(lapply(1:100, function(i){eps = rnorm(1,sd=10);keyval(i, list(x=c(i,i+eps),y=2*(eps > 0) - 1))}),"/user/hadoop/logreg")
put: org.apache.hadoop.security.AccessControlException: Permission denied: user=root, access=WRITE, inode="hadoop":hadoop:supergroup:rwxr-xr-x
[1] "/user/hadoop/logreg"

The other set of commands also gave the same error as the first when run as root:

> a <- from.dfs("/user/hadoop/regData/regTestData.csv")
get: org.apache.hadoop.security.AccessControlException: Permission denied: user=root, access=EXECUTE, inode="regData":hadoop:supergroup:rw-r--r--
Error in if (file.info(tmp)[1, "isdir"]) { :
missing value where TRUE/FALSE needed

Still clinging to the idea that the issue may be due to permissions, I checked the permissions on the files involved in both sets of commands. However, much to my chagrin, they were both the same (discovered using hadoop fs -ls):

-rw-r--r-- 1 hadoop supergroup 4698 2011-10-20 11:16 /user/hadoop/logreg
-rw-r--r-- 1 hadoop supergroup 15226 2011-10-20 11:13 /user/hadoop/regData

I have now come to one of the following four conclusions:

1) This is due to a very stupid error on my part
2) This is due to some incompatibility issues between hadoop and RHadoop
3) I should be specifying some parameters that I am not (a la the mapred list in Rhipe)
4) None of the above, this is just plain beyond me

Therefore, I now find myself posting here what's ultimately quite a long post, in the hope that someone will know exactly (or at least vaguely) what's going wrong and point me in the right direction.

Thanks,

Krish.