使用 Apache Arrow 加速 R 和 Apache Spark
已发布 2019年1月25日
作者 Javier Luraschi
Javier Luraschi 是 RStudio 的软件工程师
目前在 sparklyr 和 SparkR 项目中积极开发对 Apache Spark 中 R 的 Apache Arrow 支持。这篇文章探讨了在使用 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 也可以预期类似的改进。 具有 1000 万个数字行的本地 Spark 连接和数据帧初始化如下
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
收集
类似地，arrow 与 sparklyr 现在可以避免在从 Spark 收集数据到 R 时反序列化 R 中的数据。 这些改进不像复制数据那么重要，因为 sparklyr 已经以列式格式收集数据。
以下基准测试从 Spark 中收集 1000 万行到 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
转换
今天，使用 R 函数自定义转换数据是通过将数据以行格式从 Spark 通过套接字连接移动到 R 进程中来在 sparklyr 中执行的，以行格式传输数据效率低下，因为需要为每一行反序列化多种数据类型，然后将数据转换为列式格式（R 最初设计为使用列式数据），一旦 R 完成此计算，数据再次转换为行格式，逐行序列化，然后通过套接字连接发送回 Spark。
通过在 sparklyr 中添加对 arrow 的支持，它使 Spark 在 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
可以在 sparklyr /rstudio/sparklyr/pull/1611 和 SparkR /apache/spark/pull/22954 下找到其他基准测试和微调参数。 期待将此功能带给 Spark、Arrow 和 R 社区。