我是 Spark、SparkR 以及所有 HDFS 相关技术的新手。我最近安装了 Spark 1.5.0 并使用 SparkR 运行一些简单的代码:
Sys.setenv(SPARK_HOME="/private/tmp/spark-1.5.0-bin-hadoop2.6")
.libPaths("/private/tmp/spark-1.5.0-bin-hadoop2.6/R/lib")
require('SparkR')
require('data.table')
sc <- sparkR.init(master="local")
sqlContext <- sparkRSQL.init(sc)
hiveContext <- sparkRHive.init(sc)
n = 1000
x = data.table(id = 1:n, val = rnorm(n))
Sys.time()
xs <- createDataFrame(sqlContext, x)
Sys.time()
代码立即执行。但是,当我将其更改为
n = 1000000
时大约需要 4 分钟(两次 Sys.time()
调用之间的时间)。当我在端口 :4040 上的控制台中检查这些作业时, n = 1000
的作业持续时间为 0.2 秒,作业为 n = 1000000
0.3 秒。难道我做错了什么?
最佳答案
你没有做任何特别错误的事情。这只是不同因素组合的结果:
createDataFrame
目前(Spark 1.5.1)实现速度很慢。这是 SPARK-8277 中描述的已知问题. data.table
不兼容. 在解决 SPARK-8277 之前,您无能为力,但您可以尝试两种选择:
data.frame
而不是 data.table
.使用航类数据集(227496 行,14 列):df <- read.csv("flights.csv")
microbenchmark::microbenchmark(createDataFrame(sqlContext, df), times=3)
## Unit: seconds
## expr min lq mean median
## createDataFrame(sqlContext, df) 96.41565 97.19515 99.08441 97.97465
## uq max neval
## 100.4188 102.8629 3
与
data.table
相比dt <- data.table::fread("flights.csv")
microbenchmark::microbenchmark(createDataFrame(sqlContext, dt), times=3)
## Unit: seconds
## expr min lq mean median
## createDataFrame(sqlContext, dt) 378.8534 379.4482 381.2061 380.043
## uq max neval
## 382.3825 384.722 3
spark-csv
将数据直接加载到 Spark DataFrame 而不与 R 直接交互。听起来很疯狂:dt <- data.table::fread("flights.csv")
write_and_read <- function() {
write.csv(dt, tempfile(), row.names=FALSE)
read.df(sqlContext, "flights.csv",
source = "com.databricks.spark.csv",
header = "true",
inferSchema = "true"
)
}
## Unit: seconds
## expr min lq mean median
## write_and_read() 2.924142 2.959085 2.983008 2.994027
## uq max neval
## 3.01244 3.030854 3
我不确定首先将可以在 R 中处理的数据推送到 Spark 是否真的有意义,但不要纠结于此。
编辑 :
此问题应由 SPARK-11086 解决在 Spark 1.6.0 中。
关于r - createDataFrame 中的 SparkR 瓶颈?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/32888142/