多进程而不是R中的for循环

Question

多进程而不是R中的for循环

Dan*_*mes 5 windows parallel-processing for-loop r arima

我想跑for loop进去parallel process。我的代码结果for loop R很符合我的口味，但将其应用到非常huge data因此，执行的时间很慢。

library(forecast)
library(dplyr)
arima_order_results = data.frame()
seed_out2 <- c(1, 16, 170, 178, 411, 630, 661, 1242, 1625, 1901, 1926, 1927, 1928, 2170, 2779, 3687, 4139, 4583, 4825, 4828, 4829, 4827, 5103, 5211, 5509, 5561, 5569, 5679, 6344, 6490, 6943, 6944, 6945, 6946, 6948, 6950, 6951, 6952)
for (my_seed in seed_out2){
  set.seed(my_seed)
  ar1 <- arima.sim(n = 100, model=list(ar = 0.8, order = c(1, 0, 0)), sd = 1)
  ar2 <- auto.arima(ar1, ic = "aicc")
  arr <- as.data.frame(t(ar2$coef))
  if(substr(as.character(arr[1]), 1, 5) == "0.800") {

    arr <- cbind(data.frame(seed=my_seed),arr)
    print(arr)

    arima_order_results = bind_rows(arima_order_results,arr)
    # write.csv(my_seed, paste0(arr, ".csv"), row.names = FALSE)

  } #else print("NOT AVAILABLE")
}

Run Code Online (Sandbox Code Playgroud)

结果

#  seed       ar1
#1  170 0.8006368
#  seed       ar1
#1  411 0.8004152
#  seed       ar1
#1  630 0.8008459
#  seed       ar1
#1  661 0.8001553
#  seed       ar1 intercept
#1 1242 0.8000623 0.8474553
#  seed       ar1
#1 1625 0.8004982
#  seed       ar1
#1 1901 0.8007815
#  seed       ar1
#1 1927 0.8004587
#  seed       ar1
#1 2170 0.8003091
#  seed       ar1
#1 2779 0.8008643
#:
#:
#:
#seed      ar1
#1 5679 0.800689
#  seed     ar1 intercept
#1 6344 0.80004 0.9800426
#  seed       ar1
#1 6490 0.8004093
#  seed       ar1
#1 6948 0.8006992

Run Code Online (Sandbox Code Playgroud)

我想要的是

我想要一个并行进程，它会同时用完我的四个处理器，以便job execution will be fast when I apply it to 在处理大量数据时得到相同的结果。

看看我尝试了什么

library(parallel)    
library(foreach)
library(forecast)
library(dplyr)
library(doSNOW)
cl <- parallel::makeCluster(detectCores(), type = "SOCK")   
doSNOW::registerDoSNOW(cl)
arima_order_results = data.frame()
seed_out2 <- c(1, 16, 170, 178, 411, 630, 661, 1242, 1625, 1901, 1926, 1927, 1928, 2170, 2779, 3687, 4139, 4583, 4825, 4828, 4829, 4827, 5103, 5211, 5509, 5561, 5569, 5679, 6344, 6490, 6943, 6944, 6945, 6946, 6948, 6950, 6951, 6952)
lst_out <- foreach::foreach(my_seed = seq_along(seed_out2), .packages = c("dplyr", "forecast") ) %dopar% {
  set.seed(my_seed)
  ar1 <- arima.sim(n = 100, model=list(ar = 0.8, order = c(1, 0, 0)), sd = 1)
  ar2 <- auto.arima(ar1, ic = "aicc")
  arr <- as.data.frame(t(ar2$coef))
  if(substr(as.character(arr[1]), 1, 5) == "0.800") {

    arr <- cbind(data.frame(seed=my_seed),arr)
    print(arr)

    arima_order_results = bind_rows(arima_order_results,arr)
    # write.csv(my_seed, paste0(arr, ".csv"), row.names = FALSE)

  }
}

Run Code Online (Sandbox Code Playgroud)

查看我的试用结果

#>lst_out
#[[1]]
#NULL

#[[2]]
#NULL

#[[3]]
#NULL

#[[4]]
#NULL
#:
#:
#:
#[[36]]
#NULL

#[[37]]
#NULL

#[[38]]
#NULL

Run Code Online (Sandbox Code Playgroud)

我是在windows上操作的。

伊迪丝

我希望以这样的方式修改 @jay.sf 答案，使其包含在像我下面提供的函数这样的函数中。

FUN1 <- function(n, ar, sd, arr, R, FUN2){
  FUN2 <- function(i, n, ar, sd, arr) {
    set.seed(i)
    ar1 <- arima.sim(n=n, model=list(ar=ar, order=c(1, 0, 0)), sd=sd)
    ar2 <- auto.arima(ar1, ic="aicc")
    (cf <- ar2$coef)
    if (length(cf) == 0) {
    rep(NA, 2)
    }
    else if (all(grepl(c("ar1|intercept"), names(cf))) &  ## using `grepl`
             substr(cf["ar1"], 1, 5) %in% "arr") { 
      c(cf, seed=i)
    }
    else {
      rep(NA, 2)
    }
  }

  seedv <- 1:R

  library(parallel)
  cl <- makeCluster(detectCores() - 1)
  clusterExport(cl, c("FUN2"), envir=environment())
  clusterEvalQ(cl, suppressPackageStartupMessages(library(forecast)))

  res <- parLapply(cl, seedv, "FUN2")

  res1 <- res[!sapply(res, anyNA)]  ## filter out NAs

  stopCluster(cl)
}
FUN1(n = 10, ar = 0.8, sd = 1, arr = 0.800, R = 1000, FUN2 = FUN2)

Run Code Online (Sandbox Code Playgroud)

Answer 1

jay*_*.sf 3

这是我对您之前的相关问题之一给出的答案的类似方法。

FUN <- function(i) {
  set.seed(i)
  ar1 <- arima.sim(n=100, model=list(ar=0.8, order=c(1, 0, 0)), sd=1)
  ar2 <- auto.arima(ar1, ic="aicc")
  cf <- ar2$coef
  ## case handling
  if (length(cf) == 0) rep(NA, 2)  ## sometimes result is `character(0)` -> NA
  else if (substr(cf[1], 1, 5) %in% "0.800") c(cf, i)  ## hit, that's what we want
  else rep(NA, 2)  ## all other cases -> NA
}

R <- 1e3  ## this would be your 1e5
seedv <- 1:R  ## or use custom seed vector

library(parallel)
cl <- makeCluster(detectCores() - 1)  ## for all cores remove `- 1`
clusterExport(cl, c("FUN"), envir=environment())
clusterEvalQ(cl, suppressPackageStartupMessages(library(forecast)))

res <- `colnames<-`(t(parSapply(cl, seedv, "FUN")), c("cf", "seed"))

stopCluster(cl)

Run Code Online (Sandbox Code Playgroud)

结果

在结果中，我们想要过滤掉所有带有的行NA。

head(res[!is.na(res[,1]), ])
#             cf seed
# [1,] 0.8006368  170
# [2,] 0.8004152  411
# [3,] 0.8008459  630
# [4,] 0.8001553  661

Run Code Online (Sandbox Code Playgroud)

编辑

要包含auto.arima仅包含"ar1"和的组合的结果"intercept"，我们最好使用parLapply：

FUN <- function(i) {
  set.seed(i)
  ar1 <- arima.sim(n=50, model=list(ar=0.8, order=c(1, 0, 0)), sd=1)
  ar2 <- auto.arima(ar1, ic="aicc")
  (cf <- ar2$coef)
  if (length(cf) == 0) {
    rep(NA, 2)
    }
  else if (all(grepl(c("ar1|intercept"), names(cf))) &  ## using `grepl`
           substr(cf["ar1"], 1, 5) %in% "0.800") { 
    c(cf, seed=i)
    }
  else {
    rep(NA, 2)
    }
}

R <- 1e4
seedv <- 1:R

library(parallel)
cl <- makeCluster(detectCores() - 1)
clusterExport(cl, c("FUN"), envir=environment())
clusterEvalQ(cl, suppressPackageStartupMessages(library(forecast)))

res <- parLapply(cl, seedv, "FUN")

res1 <- res[!sapply(res, anyNA)]  ## filter out NAs

stopCluster(cl)

Run Code Online (Sandbox Code Playgroud)

这给出了具有不等列长度的数据框列表，我们可以merge使用Reduce.

res2 <- Reduce(function(...) merge(..., all=T), lapply(res1, function(x) as.data.frame(t(x))))

res2[order(res2$seed), c("ar1", "intercept", "seed")]  ## some ordering
#          ar1 intercept seed
# 1  0.8000531  1.335388  290
# 3  0.8002499        NA 2154
# 10 0.8005477        NA 2888
# 11 0.8006736        NA 3203
# 15 0.8009363        NA 4415
# 14 0.8008462        NA 4572
# 4  0.8003495        NA 4726
# 9  0.8005087        NA 6241
# 2  0.8001865        NA 6417
# 13 0.8008060 -1.700587 6845
# 6  0.8003977        NA 7187
# 8  0.8004316        NA 8981
# 7  0.8004268        NA 9368
# 12 0.8007281        NA 9697
# 5  0.8003903        NA 9793

Run Code Online (Sandbox Code Playgroud)

编辑2

这是一个只需要用户指定R迭代次数的函数。它在内部用于doParallel::registerDoParallel定义一个隐式集群，detectCores() - 1默认情况下使用通常的集群，但也可以由用户指定。集群将自动停止。此外，还foreach应用了循环。

library(forecast)
library(doParallel)

arimaze <- function(R, ncores=detectCores() - 1) {
  registerDoParallel(ncores)
  seedv <- 1:R
  FUN <- function(i) {
    set.seed(i)
    ar1 <- arima.sim(n=50, model=list(ar=0.8, order=c(1, 0, 0)), sd=1)
    ar2 <- auto.arima(ar1, ic="aicc")
    cf <- ar2$coef
    if (length(cf) == 0 | !(all(grepl(c("ar1|intercept"), names(cf))) &
                            substr(cf["ar1"], 1, 5) %in% "0.800")) {
      return(rep(NA, 3))
    } else {
      cf <- `length<-`(cf, 2)
      return(c(cf, seed=i))
    }
  }
  message('processing...')
  res <-
    foreach(i=seedv, .combine=rbind.data.frame, .packages='forecast') %dopar% 
    FUN(i)
  message(' done!\n')
  res <- `rownames<-`(res[rowSums(is.na(res)) == 0, ], NULL)
  stopImplicitCluster()
  return(setNames(res, c('ar', 'intercept', 'seed')))
}

Run Code Online (Sandbox Code Playgroud)

用法

r <- arimaze(1.5e4)
# processing... done!

Run Code Online (Sandbox Code Playgroud)

结果

r
#          ar intercept  seed
# 1 0.8000531  1.335388   290
# 2 0.8008060 -1.700587  6845
# 3 0.8003690 -1.443856 12137

Run Code Online (Sandbox Code Playgroud)

归档时间：	4 年，11 月前
查看次数：	269 次
最近记录：	2 年，4 月前