嵌套循环的并行处理在 R 中无法正常工作

Question

我是 R 新手，我正在尝试使用 parallel() 和 foreach() 包运行嵌套循环来提高速度，但我没有成功。 我有一个数据框，其中第一列是分类数据，其他列是标量。我想运行一个函数，例如 caret::bagFDA，使用两个预测变量和分类变量作为响应变量，以便测试所有可能的组合而不重复。 所以，我正在寻找的是“分类 ~ 标量变量 1 + 标量变量 2”，在一个嵌套循环中，所有组合都执行而不重复。

当我尝试运行传统的嵌套循环时，它工作得很好。这是一个可重现的示例：

#Install packages
install.packages("caret")
install.packages("dplyr")

start.time <- Sys.time()
#Libraries
library(caret)
library(dplyr)

#Read iris data
data(iris)
iris <- iris[c(ncol(iris), 1:(ncol(iris) - 1))]

#Sampling 50/50 for train and valid
set.seed(2000)
Train <- iris %>% group_by(Species) %>% sample_frac(.5, replace = FALSE)
Valid <- anti_join(iris, Train)


# bagFDA
BAGFDA <- data.frame(Variable_Name1 = character(0), Variable_Name2 = character(0), Accuracy = numeric(0), Kappa = numeric(0))

set.seed(3000)
for(i in 2:(ncol(Train) - 1)) {
  for (j in (i + 1):ncol(Train)) {
    tryCatch({
      formula <- as.formula(paste("as.factor(Species) ~", names(Train)[i], "+", names(Train)[j]))
      bag <- caret::bagFDA(formula, data = Train)
      bag_predict <- predict(bag, newdata = Valid)
      bag_CM <- confusionMatrix(bag_predict, Valid$Species)
      iteration_results <- data.frame(
        Variable_Name1 = names(Train)[i],
        Variable_Name2 = names(Train)[j],
        Accuracy = bag_CM$overall["Accuracy"],
        Kappa = bag_CM$overall["Kappa"]
      )
      BAGFDA <- rbind(BAGFDA, iteration_results)
      print("Good")
    }, error = function(e) {
      cat("ERROR:", conditionMessage(e), "\n")
    })
  }
}
print(BAGFDA)
end.time <- Sys.time()
time.taken <- round(end.time - start.time,2)
time.taken

我不确定这是否是运行代码的最优化方式，但它运行良好，大约花费了 4 秒。 但是，当我尝试与 parallel() 和 foreach() 包并行运行它时，我无法获得相同的结果。这是我正在尝试的代码：

#Install packages
install.packages("foreach")
install.packages("doParallel")
install.packages("caret")
install.packages("dplyr")

start.time <- Sys.time()
#Libraries
library(foreach)
library(doParallel)
library(caret)
library(dplyr)

#Read iris data
data(iris)
iris <- iris[c(ncol(iris), 1:(ncol(iris) - 1))]

#Creating clusters
cores = parallel::detectCores() - 1
cluster = parallel::makeCluster(cores, type = "PSOCK")
doParallel::registerDoParallel(cluster)
if (!foreach::getDoParRegistered()) 
{ 
  print("ERROR")
}
print(foreach::getDoParWorkers()) 

#Sampling 50/50 for train and valid
set.seed(2000)
Train <- iris %>% group_by(Species) %>% sample_frac(.5, replace = FALSE)
Valid <- anti_join(iris, Train)

# bagFDA
BAGFDA <- data.frame(Variable_Name1 = character(0), Variable_Name2 = character(0), Accuracy = numeric(0), Kappa = numeric(0))

set.seed(1001)
results <- foreach(i = 2:(ncol(Train) - 1), .combine='cbind') %:%
  foreach (j = (i + 1):ncol(Train), .combine='c') %dopar%
    tryCatch({
      formula <- as.formula(paste("as.factor(Species) ~", names(Train)[i], "+", names(Train)[j]))
      bag <- caret::bagFDA(formula, data = Train)
      bag_predict <- predict(bag, newdata = Valid)
      bag_CM <- confusionMatrix(bag_predict, Valid$Species)
      iteration_results <- data.frame(
        Variable_Name1 = names(Train)[i],
        Variable_Name2 = names(Train)[j],
        Accuracy = bag_CM$overall["Accuracy"],
        Kappa = bag_CM$overall["Kappa"]
      )
      BAGFDA <- rbind(BAGFDA, iteration_results)
      print("Good")
    }, error = function(e) {
      cat("ERROR:", conditionMessage(e), "\n")
    })

print(BAGFDA)
stopCluster(cluster)
end.time <- Sys.time()
time.taken <- round(end.time - start.time,2)
time.taken

运行没有显示任何错误，但数据帧 BAGFDA 未填充，结果始终为 NULL，并且公式始终为（未知）。 显然，我做错了什么，但我无法识别错误。请你帮助我好吗？ 我遵循本文档中提供的指南：https://cran.r-project.org/web/packages/foreach/vignettes/nested.html

谢谢！

Answer 1

试试这个：

results <- foreach(i = 2:(ncol(Train) - 1), .combine=rbind) %:%
  foreach (j = (i + 1):ncol(Train), .combine=rbind, .packages = c("caret")) %dopar% {
    tryCatch({
      formula <- as.formula(paste("Species ~", names(Train)[i], "+", names(Train)[j]))
      bag <- caret::bagFDA(formula, data = Train)
      bag_predict <- predict(bag, newdata = Valid)
      bag_CM <- confusionMatrix(bag_predict, Valid$Species)
      
      data.frame(
        Variable_Name1 = names(Train)[i],
        Variable_Name2 = names(Train)[j],
        Accuracy = bag_CM$overall["Accuracy"],
        Kappa = bag_CM$overall["Kappa"]
      )
    }, error = function(e) {
      cat("ERROR:", conditionMessage(e), "\n")
    })
  }

我相信您遇到的问题是由于工人无法使用

caret

软件包引起的。在摆弄原始代码后，我得到了这个错误：

Error in { : task 1 failed - "could not find function "confusionMatrix""

这表明这可能是问题所在。

传递

.packages = c("caret")

参数可确保该包可供作品使用。

上述代码的结果是：

           Variable_Name1 Variable_Name2  Accuracy     Kappa
Accuracy     Sepal.Length    Sepal.Width 0.7837838 0.6755275
Accuracy1    Sepal.Length   Petal.Length 0.9729730 0.9594521
Accuracy2    Sepal.Length    Petal.Width 0.9594595 0.9391614
Accuracy3     Sepal.Width   Petal.Length 0.9729730 0.9594521
Accuracy11    Sepal.Width    Petal.Width 0.9594595 0.9391614
Accuracy4    Petal.Length    Petal.Width 0.9594595 0.9391614

另请注意，我在两个

.combine=rbind

命令中都使用了

foreach()

。这意味着您不需要在输入

BAGFDA

之前创建结果数据框 (

foreach()

)，并且还可以避免在循环中调用

rbind()

，正如评论中已经指出的那样，这可能会对性能产生影响如果您正在处理更大的数据（假设您可能会将其应用于

iris

以外的其他事物）。

需要注意的一件事是

set.seed(1001)

可能不会达到预期的效果。运行以下几次，您会发现每次运行都会得到不同的随机数。原因是master中设置的种子不会传播到每个worker。

set.seed(1001)
foreach(i = 2:(ncol(Train) - 1)) %:%
  foreach (j = (i + 1):ncol(Train), .packages = c("caret")) %dopar% {
    runif(3)
  }