将 sparkr 收集到数据框中

Question

我正在将一些数据加载到 sparkR（Spark 版本 1.4.0，运行ning 在 fedora21 上），我运行一些生成三个不同数字的算法。我的算法需要一堆参数，我想运行对同一数据进行不同的参数设置。输出格式应该是一个数据框（或一个 csv 列表），其列是算法参数和我的算法计算的三个数字，即

  mypar1, mypar2, mypar3, myres1, myres2, myres3
  1       1.5     1.2     5.6      8.212  5.9
  2       1.8     1.7     5.1      7.78   8.34

将是两个不同参数设置的输出。我写了下面的脚本，它在不同的参数设置上并行化运行ning：它需要一个带有参数值的输入文件作为参数，对于上面的例子来说，它看起来像这样：

 1,1.5,1.2
 2,1.8,1.7

所以每行一个参数组合。

这是我的问题：不是每个参数设置一个，而是将所有数字合并到一个长列表中。函数 cv_spark returns a data.frame （基本上是一行）。我如何告诉 spark 将 cv_spark 的输出组合到一个数据帧中（即做类似 rbind 的事情？）或列表列表？

#!/home/myname/Spark/spark-1.4.0/bin/sparkR

library(SparkR)

sparkcontext <- sparkR.init("local[3]","cvspark",sparkEnvir=list(spark.executor.memory="1g"))

cv_spark <- function(indata) {
   cv_params <- strsplit(indata, split=",")[[1]]
   param.par1 = as.integer(cv_params[1])
   param.par2 = as.numeric(cv_params[2])
   param.par3 = as.numeric(cv_params[3])
   predictions <- rep(NA, 1)
   ## here I run some calculation on some data that I load to my SparkR session, 
   ## but for illustration purpose I'm just filling up with some random numbers
   mypred = base:::sample(seq(5,10,by=0.01),3)
   predictions <- cbind(param.par1, param.par2, param.par3,mypred[1],mypred[2],mypred[3])
   return(as.data.frame(predictions))
}

args <- commandArgs(trailingOnly=TRUE)
print(paste("args ", args))
cvpar = readLines(args[[1]])

rdd <- SparkR:::parallelize(sparkcontext, coll=cvpar, numSlices=4)
myerr <- SparkR:::flatMap(rdd,cv_spark)
output <- SparkR:::collect(myerr)
print("final output")
print(output)

outfile = "spark_output.csv"
write.csv(output,outfile,quote=FALSE,row.names=FALSE)

Answer 1

我设法通过使用 flatMapValues 而不是 flatMap 并通过创建 (key, value) 对我的各种参数设置（基本上关键是我的行号参数输入文件和值是该行上的参数）。然后我调用 reduceByKey ，它基本上每个键保存一行。修改后的脚本如下所示：

#!/home/myname/Spark/spark-1.4.0/bin/sparkR

library(SparkR)

sparkcontext <- sparkR.init("local[4]","cvspark",sparkEnvir=list(spark.executor.memory="1g"))

cv_spark <- function(indata) {
   cv_params <- unlist(strsplit(indata[[1]], split=","))
   param.par1 = as.integer(cv_params[1])
   param.par2 = as.numeric(cv_params[2])
   param.par3 = as.integer(cv_params[3])
   predictions <- rep(NA, 1)
   ## here I run some calculation on some data that I load to my SparkR session, 
   ## but for illustration purpose I'm just filling up with some random numbers
   mypred = base:::sample(seq(5,10,by=0.01),3)
   predictions <- cbind(param.par1, param.par2, param.par3,mypred[1],mypred[2],mypred[3])
   return(as.data.frame(predictions))
}

args <- commandArgs(trailingOnly=TRUE)
print(paste("args ", args))
cvpar = readLines(args[[1]])
## Creates (key, value) pairs
cvpar <- Map(list,seq(1,length(cvpar)),cvpar)

rdd <- SparkR:::parallelize(sparkcontext, coll=cvpar, numSlices=1)
myerr <- SparkR:::flatMapValues(rdd,cv_spark)
myerr <- SparkR:::reduceByKey(myerr,"c", 2L)
output <- SparkR:::collect(myerr)

myres <- sapply(output,`[`,2)
df_res <- do.call("rbind",myres)
colnames(df_res) <- c("Element","sigdf","sigq","err","err.sse","err.mse")

outfile = "spark_output.csv"
write.csv(df_res,outfile,quote=FALSE,row.names=FALSE)

这按预期工作，即输出是一个数据框（或 csv 文件），其行数与上述脚本的输入文件中的行数相同（即不同参数值配置的数量），但也许有一种更有效的方法。

将 sparkr 收集到数据框中

collect sparkr into dataframe

r

apache-spark

sparkr