Как раскошелиться / распараллелить процесс в purrr::pmap

Question

У меня есть следующий код, который выполняет последовательную обработку с purr::pmap

library(tidyverse)

set.seed(1)
params <- tribble(
  ~mean, ~sd, ~n,
  5,     1,  1,
  10,     5,  3,
  -3,    10,  5
)
params %>% 
  pmap(rnorm)
#> [[1]]
#> [1] 4.373546
#> 
#> [[2]]
#> [1] 10.918217  5.821857 17.976404
#> 
#> [[3]]
#> [1]   0.2950777 -11.2046838   1.8742905   4.3832471   2.7578135

Как я могу распараллелить (разветвить) описанный выше процесс, чтобы он работал быстрее и давал идентичный результат?

Здесь я использую rnorm для иллюстрации, на самом деле у меня есть функция, которая выполняет тяжелую работу. Это требует распараллеливания.

Я открыт для решения, не связанного с мурлыканием (без тидиверса), при условии, что оно дает идентичный результат, учитывая rnorm функция и params в качестве ввода.

Answer 1

Короче говоря: "параллель" pmap()", позволяющий подобный синтаксис pmap(), может выглядеть так: lift(mcmapply)() или же lift(clusterMap)(),

---

Если вы не используете Windows, вы можете:

library(parallel)

# forking

set.seed(1, "L'Ecuyer")
params %>% 
  lift(mcmapply, mc.cores = detectCores() - 1)(FUN = rnorm)

# [[1]]
# [1] 4.514604
# 
# [[2]]
# [1] 0.7022156 0.8734875 5.0250478
# 
# [[3]]
# [1]   8.7704060  11.7217925 -12.8776289 -10.7466152   0.5177089

редактировать

Вот "более чистый" вариант, который должен больше походить на использование pmap:

nc <- max(parallel::detectCores() - 1, 1L)

par_pmap <- function(.l, .f, ..., mc.cores = getOption("mc.cores", 2L)) {
  do.call(
    parallel::mcmapply, 
    c(.l, list(FUN = .f, MoreArgs = list(...), SIMPLIFY = FALSE, mc.cores = mc.cores))
  )
}

f <- function(n, mean, sd, ...) rnorm(n, mean, sd) 

params %>% 
  par_pmap(f, some_other_arg_to_f = "foo", mc.cores = nc)

---

Если вы используете Windows (или любую другую ОС), вы можете:

library(parallel)

# (Parallel SOCKet cluster)

cl <- makeCluster(detectCores() - 1)

clusterSetRNGStream(cl, 1)
params %>% 
  lift(clusterMap, cl = cl)(fun = rnorm)

# [[1]]
# [1] 5.460811
# 
# [[2]]
# [1] 7.573021 6.870994 5.633097
# 
# [[3]]
# [1] -21.595569 -21.253025 -12.949904  -4.817278  -7.650049

stopCluster(cl)

---

Если вы более склонны использовать foreach, вы могли бы:

library(doParallel)

# (fork by default on my Linux machine, should PSOCK by default on Windows)

registerDoParallel(cores = detectCores() - 1)

set.seed(1, "L'Ecuyer")
lift(foreach)(params) %dopar%
  rnorm(n, mean, sd)

# [[1]]
# [1] 4.514604
# 
# [[2]]
# [1] 0.7022156 0.8734875 5.0250478
# 
# [[3]]
# [1]   8.7704060  11.7217925 -12.8776289 -10.7466152   0.5177089

stopImplicitCluster()