Nút cổ chai SparkR trong createDataFrame?

Question

Tôi mới sử dụng Spark, SparkR và thường là tất cả các công nghệ liên quan đến HDFS. Tôi đã cài đặt Spark 1.5.0 gần đây và chạy một số mã đơn giản với 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() 

Mã sẽ thực thi ngay lập tức. Tuy nhiên, khi tôi đổi nó thành n = 1000000, mất khoảng 4 phút (thời gian giữa hai cuộc gọi Sys.time()). Khi tôi kiểm tra các công việc này trong bảng điều khiển trên cổng: 4040, công việc cho n = 1000 có thời lượng 0,2 giây và công việc cho n = 1000000 0,3 giây. Tôi có làm điều gì sai?

Answer 1

Bạn không làm gì đặc biệt sai. Đó chỉ là ảnh hưởng của sự kết hợp các yếu tố khác nhau:

1. createDataFrame vì hiện tại (Spark 1.5.1) được triển khai chậm. Đây là một vấn đề đã biết được mô tả trong SPARK-8277.
2. Triển khai hiện tại không hoạt động tốt với data.table.
3. Cơ sở R tương đối chậm. Người thông minh nói rằng đó là một tính năng không phải là một lỗi nhưng nó vẫn là một cái gì đó để xem xét.

Cho đến SPARK-8277 đã được giải quyết không có nhiều bạn có thể làm nhưng có hai lựa chọn, bạn có thể thử:

• sử dụng đồng bằng cũ data.frame thay vì data.table. Sử dụng các chuyến bay tập dữ liệu (227.496 hàng, 14 cột):

  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 
  

  so với 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 
  

• Viết vào đĩa và sử dụng spark-csv để tải dữ liệu trực tiếp đến Spark DataFrame mà không tương tác trực tiếp với R. Như điên như âm thanh:

  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 
  

tôi không tái đồng minh chắc chắn nếu thực sự nó có ý nghĩa để đẩy dữ liệu có thể được xử lý trong R để Spark ở nơi đầu tiên nhưng cho phép không nằm trên đó.

Sửa:

Vấn đề này nên được giải quyết bằng SPARK-11086 trong Spark 1.6.0.