用doMC代替并行plyr

Question

考虑一下data.frame上的标准分组操作：

library(plyr)
library(doMC)
library(MASS) # for example

nc <- 12
registerDoMC(nc)

d <- data.frame(x = c("data", "more data"), g = c("group1", "group2"))
y <- "some global object"

res <- ddply(d, .(g), function(d_group) {
   # slow, complicated operations on d_group
}, .parallel = FALSE)

简单地编写.parallel = TRUE来利用多核设置是很简单的。这是我最喜欢的功能之一。

但是，随着plyr被废弃(我认为)并基本上被dplyr、purrr等所取代，并行处理的解决方案变得更加冗长：

library(dplyr)
library(multidplyr)
library(parallel)
library(MASS) # for example

nc <- 12

d <- tibble(x = c("data", "more data"), g = c("group1", "group2"))
y <- "some global object"

cl <- create_cluster(nc)
set_default_cluster(cl)
cluster_library(cl, packages = c("MASS"))
cluster_copy(cl, obj = y)

d_parts <- d %>% partition(g, cluster = cl)
res <- d_parts %>% collect() %>% ungroup()

rm(d_parts)
rm(cl)

可以想象，考虑到循环中需要的每个包和对象需要多长时间，需要自己的cluster_*命令将其复制到节点上。非并行化的plyr到dplyr转换只是一个简单的dplyr::group_by结构，不幸的是，没有简单的方法在其上启用并行处理。所以，我的问题是：

• 这实际上是将我的代码从plyr转换为dplyr的首选方法吗？
• 在plyr的幕后发生了什么样的魔术，使得打开并行处理变得如此容易？这种功能为什么特别难以添加到dplyr中，这就是为什么它还不存在的原因吗？
• 我的两个例子在代码的执行方式上有根本的不同吗？

Answer 1

1. 我认为没有一种真正的“首选”方法可以将{plyr}代码转换为{dplyr}。
2. 在注释@Aurèle中，在描述{plyr}和{doMC}之间的连接方面，我做得比以往任何时候都要好。有一件事发生了，那就是激励机制发生了一些变化。{doMC}来自革命分析(自微软购买)。但开发dplyr的哈德利目前在RStudio工作。这两家公司在IDE领域展开竞争。所以，他们的软件包设计得不适合一起玩，这可能是很自然的。我所看到的对RStudio的强大支持的唯一形式是{sparklyr}，它们使其相对“容易”设置。但是，我不能真正地建议与火花进行并行处理的单一机器。
3. @Aurèle再次出色地解释了执行方面的差异。新代码使用PSOCK集群，旧代码使用分叉。分叉使用副本上的写入模式访问RAM，因此并行进程可以从访问相同的数据开始立即post叉。PSOCK集群就像生成新的R副本--它们必须加载库并接收数据的显式副本。

你可以用这样的模式。

library(dplyr)
library(purrr)
library(future)
plan(multicore)
options(mc.cores = availableCores())
d <- data.frame(x = 1:8, g = c("group1", "group2", "group3", "group4"))
y <- "some global object"


split(d, d$g) %>% 
  map(~ future({Sys.sleep(5);mean(.x$x)})) %>% 
  map_df(~value(.x))

..。用一些技巧对map_df步骤做一些并行处理。注意，在{purrr}下，~是匿名函数语法，其中.x是已映射到的值。

如果您喜欢危险地生活，则可以通过使用{purrr}中的私有方法来创建类似的版本，而不必使用{未来}。

mcmap <- function(.x, .f, ...) {
  .f <- as_mapper(.f, ...)
  mclapply(.x, function(.x) {
    force(.f)
    .Call(purrr:::map_impl, environment(), ".x", ".f", "list")
  }) %>%
    map(~ .x[[1]])
}