createDataFrame 中的 SparkR 瓶颈?

SparkR bottleneck in createDataFrame?

我是 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 秒。我做错了什么吗?

你没有做错什么。这只是不同因素组合的结果:

  1. createDataFrame 目前(Spark 1.5.1)实施速度很慢。这是 SPARK-8277.
  2. 中描述的已知问题
  3. 当前的实现与 data.table 不兼容。
  4. Base R 相对较慢。聪明人说这是一个功能而不是错误,但它仍然需要考虑。

在解决 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 中解决。