我对 SparkR(以及一般并行化)非常陌生。我在本地运行 SparkR(我知道这不是 spark 的正确用法,但我才刚刚开始)并且我尝试用 sparkR 重写我的代码的某些部分,尽管 collect 通过增加数量给我以下错误样本为(少量样本没有错误):
Error in unserialize(obj) :
ReadItem: unknown type 0, perhaps written by later version of R
Calls: assetForecast ... convertJListToRList -> lapply -> lapply -> FUN -> unserialize
Execution halted
另一个可能是因为我的内存不足的错误是:
heap memory error (trying increasing JVM memory & driver memory did not help)
我将不胜感激有关第一个错误的任何帮助(我发布了第二个错误,因为我认为它们可能以某种方式相关,即使我通过为 numSlices 并行化设置不同的值来获得它们)。我认为第一个可能是 spark、sparkR 和 R 之间的版本不兼容导致了这个序列化问题。我尝试安装不同的版本,但很快就解决了依赖问题。
这是一个示例脚本,它模拟我在 SparkR 中所做的事情(为 input.len > 950 生成错误):
library(SparkR) # load sparkR library
sc <- sparkR.init() ## initialize the sparkR
input.len <- 8000 # size of the input
num.slice <- 2 # number of slices for parallelize function
## Define a few functions to simulate actual calculations
latemail <- function(N, st="2012/01/01", et="2015/12/31") {
## create random date of length N
st <- as.POSIXct(as.Date(st))
et <- as.POSIXct(as.Date(et))
dt <- as.numeric(difftime(et,st,unit="sec"))
ev <- sort(runif(N, 0, dt))
rt <- st + ev
}
encode <- function(ele1, ele2) {
## concatenate ele1 and ele2, seperated by %
return (paste(toString(ele1), toString(ele2), sep = "%"))
}
decode <- function(coded) {
## separate input string by %
idx <- regexpr("%", coded)[1]
ele1 <- as.numeric(substr(coded, 1, idx-1))
ele2 <- substr(coded, idx + 1, nchar(coded))
return (list(ele1, ele2))
}
fakeFun <- function(asset.age, asset.year) {
## fake function to simulate my actual function
return (as.list(rep(asset.age, 10)))
}
wrapperFun <- function(x) {
asset.age <- decode(x)[[1]]
asset.y <- decode(x)[[1]]
df <- fakeFun(asset.age, asset.y)
return (df)
}
## Start of calculations with SparkR
calc.ts <- latemail(input.len) ## create fake years
asset.ages <- runif(input.len) * 10 ## create fake ages
paired <- list()
for (i in 1:length(asset.ages)) {
## keep information of both years and ages in one vector
## using encode function
paired[[length(paired) + 1]] <- encode(asset.ages[[i]], calc.ts[[i]])
}
rdd.paired <- parallelize(sc, paired, numSlices = num.slice)
rdd.df <- lapply(rdd.paired, wrapperFun)
rdd.list <- collect(rdd.df)
print(rdd.list)
sparkR.stop()
以下是完整的错误报告:
for numSlice = 5 in parallelize function:
> rdd.list <- collect(rdd.df)
15/07/22 17:20:40 INFO RRDD: Times: boot = 0.434 s, init = 0.015 s, broadcast = 0.000 s, read-input = 0.003 s, compute = 0.200 s, write-output = 0.004 s, total = 0.656 s
15/07/22 17:20:41 INFO RRDD: Times: boot = 0.010 s, init = 0.017 s, broadcast = 0.000 s, read-input = 0.003 s, compute = 0.193 s, write-output = 0.004 s, total = 0.227 s
15/07/22 17:20:41 INFO RRDD: Times: boot = 0.010 s, init = 0.013 s, broadcast = 0.001 s, read-input = 0.002 s, compute = 0.191 s, write-output = 0.003 s, total = 0.220 s
15/07/22 17:20:41 INFO RRDD: Times: boot = 0.010 s, init = 0.011 s, broadcast = 0.000 s, read-input = 0.002 s, compute = 0.191 s, write-output = 0.004 s, total = 0.218 s
15/07/22 17:20:41 INFO RRDD: Times: boot = 0.014 s, init = 0.015 s, broadcast = 0.000 s, read-input = 0.003 s, compute = 0.213 s, write-output = 0.004 s, total = 0.249 s
Error in unserialize(obj) :
ReadItem: unknown type 0, perhaps written by later version of R
Calls: collect ... convertJListToRList -> lapply -> lapply -> FUN -> unserialize
Execution halted
for numSlice = 6 in parallelize function
15/07/22 17:18:52 WARN TaskSetManager: Lost task 2.0 in stage 0.0 (TID 2, localhost): java.lang.OutOfMemoryError: Java heap space
edu.berkeley.cs.amplab.sparkr.RRDD.readData(RRDD.scala:258)
edu.berkeley.cs.amplab.sparkr.RRDD.readData(RRDD.scala:243)
edu.berkeley.cs.amplab.sparkr.BaseRRDD.read(RRDD.scala:200)
edu.berkeley.cs.amplab.sparkr.BaseRRDD$$anon$1.next(RRDD.scala:70)
scala.collection.Iterator$class.foreach(Iterator.scala:727)
edu.berkeley.cs.amplab.sparkr.BaseRRDD$$anon$1.foreach(RRDD.scala:66)
scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:48)
scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:103)
scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:47)
scala.collection.TraversableOnce$class.to(TraversableOnce.scala:273)
edu.berkeley.cs.amplab.sparkr.BaseRRDD$$anon$1.to(RRDD.scala:66)
scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:265)
edu.berkeley.cs.amplab.sparkr.BaseRRDD$$anon$1.toBuffer(RRDD.scala:66)
scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:252)
edu.berkeley.cs.amplab.sparkr.BaseRRDD$$anon$1.toArray(RRDD.scala:66)
org.apache.spark.rdd.RDD$$anonfun$16.apply(RDD.scala:774)
org.apache.spark.rdd.RDD$$anonfun$16.apply(RDD.scala:774)
org.apache.spark.SparkContext$$anonfun$runJob$4.apply(SparkContext.scala:1121)
org.apache.spark.SparkContext$$anonfun$runJob$4.apply(SparkContext.scala:1121)
org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:62)
org.apache.spark.scheduler.Task.run(Task.scala:54)
org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:177)
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
java.lang.Thread.run(Thread.java:745)
15/07/22 17:18:52 ERROR TaskSetManager: Task 2 in stage 0.0 failed 1 times; aborting job
Error in readTypedObject(con, type) :
Unsupported type for deserialization
Calls: collect ... callJMethod -> invokeJava -> readObject -> readTypedObject
Execution halted
我的 SparkR 安装真的有问题吗?如果是,它如何针对少量样本运行?
非常感谢