简单 SparkR 1.6 示例

Question

我正在努力让一个简单的 sparkR 示例启动并运行，但无法让它工作。我要做的就是获取一个列表，将其并行化，然后在 SparkR 中运行一个函数。我还没有找到一个简单的例子来说明在 spark 1.6 中执行此操作的方法——似乎有很多方法可以做到这一点，但它们似乎都不起作用。我正在尝试在 sparkR 控制台中运行它：

fxntest<- function(i){
  m = matrix(runif(4000*4000),4000)
  return(system.time(m %*% m))
}

a<-c(1,1,1,1,1,3)
b<-as.list(a)

sdf <- createDataFrame(sqlContext, b)
testtimes <- lapply(sdf, fxntest)
testtimelist <- collect(testtimes)
testtimelist

这是我在调用 lapply 时遇到的错误：

Error in as.list.default(X) : 
  no method for coercing this S4 class to a vector

我正在寻找的是一个非常简单的例子，它可以让它工作——我已经大量使用了 pyspark，但无法使用 R 来实现它。

Answer 1

createDataFrame 已重命名为 as.DataFrame，因为 1.6。

这可以正确地在 R 中创建 Spark DataFrame。

sdf <- as.DataFrame(sqlContext, list(1,1,1,1,1,3))

但是，那不是 R 对象。内置的 R lapply 适用于常规列表、数组、向量等。

在 Spark 中运行它不会获得任何收益。您正在执行的计算不是跨集群并行化的。矩阵乘法是在一台机器上完成的。并且根本不使用 Spark 库。

这里是工作代码，例如

fxntest<- function(i){
  m = matrix(runif(4000*4000),4000)
  return(system.time(m %*% m))
}

testtimes <- lapply(list(1,1,1,1,1,3), fxntest)

注意：spark.lapply 是在 Spark 2.x 中添加的，如果这是您尝试使用的功能。