使用 Apache Arrow 加速 R 和 Apache Spark

发布 2019年1月25日
作者 Javier Luraschi

Apache Spark 中对 Apache Arrow 的 R 支持目前正在 sparklyr 和 SparkR 项目中积极开发。本文探讨了使用 R 和 Apache Spark、Arrow 以及 sparklyr 时实现的一些早期但前景广阔的性能改进。

设置

由于这项工作正在积极开发中，请按如下方式从 GitHub 安装 sparklyr 和 arrow

devtools::install_github("apache/arrow", subdir = "r", ref = "apache-arrow-0.12.0")
devtools::install_github("rstudio/sparklyr", ref = "apache-arrow-0.12.0")

在此基准测试中，我们将使用 dplyr，但使用 DBI 或 sparklyr 中的 Spark DataFrames 也能获得类似的改进。本地 Spark 连接和包含 1000 万个数字行的 dataframe 初始化如下

library(sparklyr)
library(dplyr)

sc <- spark_connect(master = "local", config = list("sparklyr.shell.driver-memory" = "6g"))
data <- data.frame(y = runif(10^7, 0, 1))

复制

目前，使用 sparklyr 将数据复制到 Spark 是通过将数据从 R 持久化到磁盘并从 Spark 中读回的方式进行的。这原本是为了小数据集设计的，因为有更好的工具可以将数据传输到分布式存储系统。然而，许多用户要求支持以更快的速度将更多数据传输到 Spark。

使用 arrow 和 sparklyr，我们可以直接将数据从 R 传输到 Spark，而无需在 R 中序列化此数据或将其持久化到磁盘。

以下示例使用 sparklyr 在启用和未启用 arrow 的情况下将 1000 万行数据从 R 复制到 Spark，使用 arrow 后性能提升了近 16 倍。

此基准测试使用 microbenchmark R 包，该包多次运行代码，提供总执行时间统计信息，并绘制每次执行时间以了解每次迭代的分布。

microbenchmark::microbenchmark(
  setup = library(arrow),
  arrow_on = {
    sparklyr_df <<- copy_to(sc, data, overwrite = T)
    count(sparklyr_df) %>% collect()
  },
  arrow_off = {
    if ("arrow" %in% .packages()) detach("package:arrow")
    sparklyr_df <<- copy_to(sc, data, overwrite = T)
    count(sparklyr_df) %>% collect()
  },
  times = 10
) %T>% print() %>% ggplot2::autoplot()

 Unit: seconds
      expr       min        lq       mean    median         uq       max neval
  arrow_on  3.011515  4.250025   7.257739  7.273011   8.974331  14.23325    10
 arrow_off 50.051947 68.523081 119.946947 71.898908 138.743419 390.44028    10

Copying data with R into Spark with and without Arrow

收集

同样，arrow 与 sparklyr 现在可以在将数据从 Spark 收集到 R 时避免在 R 中反序列化数据。这些改进不如复制数据那样显著，因为 sparklyr 已经以列式格式收集数据。

以下基准测试将 1000 万行数据从 Spark 收集到 R，并表明 arrow 可以带来 3 倍的改进。

microbenchmark::microbenchmark(
  setup = library(arrow),
  arrow_on = {
    collect(sparklyr_df)
  },
  arrow_off = {
    if ("arrow" %in% .packages()) detach("package:arrow")
    collect(sparklyr_df)
  },
  times = 10
) %T>% print() %>% ggplot2::autoplot()

Unit: seconds
      expr      min        lq      mean    median        uq       max neval
  arrow_on 4.520593  5.609812  6.154509  5.928099  6.217447  9.432221    10
 arrow_off 7.882841 13.358113 16.670708 16.127704 21.051382 24.373331    10

Collecting data with R from Spark with and without Arrow

转换

如今，使用 R 函数对数据进行自定义转换是在 sparklyr 中完成的，方法是将行格式的数据从 Spark 通过套接字连接移动到 R 进程中，以行格式传输数据效率低下，因为需要对每行中的多种数据类型进行反序列化，然后将数据转换为列格式（R 最初设计为使用列式数据），一旦 R 完成此计算，数据再次转换为行格式，逐行序列化，然后通过套接字连接发送回 Spark。

通过在 sparklyr 中添加对 arrow 的支持，Spark 可以并行执行行格式到列格式的转换。然后数据通过套接字传输，但不会发生自定义序列化。所有 R 进程需要做的就是将这些数据从套接字复制到其堆中，对其进行转换，然后将其复制回套接字连接。

以下示例在启用和未启用 arrow 的情况下转换 10 万行数据，arrow 使使用 R 函数的转换速度快了近 41 倍。

microbenchmark::microbenchmark(
  setup = library(arrow),
  arrow_on = {
    sample_n(sparklyr_df, 10^5) %>% spark_apply(~ .x / 2) %>% count()
  },
  arrow_off = {
    if ("arrow" %in% .packages()) detach("package:arrow")
    sample_n(sparklyr_df, 10^5) %>% spark_apply(~ .x / 2) %>% count()
  },
  times = 10
) %T>% print() %>% ggplot2::autoplot()

Unit: seconds
      expr        min         lq       mean     median         uq        max neval
  arrow_on   3.881293   4.038376   5.136604   4.772739   5.759082   7.873711    10
 arrow_off 178.605733 183.654887 213.296238 227.182018 233.601885 238.877341    10

Transforming data with R in Spark with and without Arrow

更多基准测试和微调参数可以在 sparklyr /rstudio/sparklyr/pull/1611 和 SparkR /apache/spark/pull/22954 下找到。期待将此功能带给 Spark、Arrow 和 R 社区。