标签: parallel-foreach

我有一个for循环,它是这样的:

for (i=1:150000) {
   tempMatrix = {}
   tempMatrix = functionThatDoesSomething() #calling a function
   finalMatrix =  cbind(finalMatrix, tempMatrix)

}

你能告诉我如何让它平行吗？

我在网上尝试了这个例子,但我不确定语法是否正确.它也没有太多提高速度.

finalMatrix = foreach(i=1:150000, .combine=cbind) %dopar%  {
   tempMatrix = {}
   tempMatrix = functionThatDoesSomething() #calling a function

   cbind(finalMatrix, tempMatrix)

}

如果我在foreach... %dopar%没有注册集群的情况下运行,foreach会发出警告,并按顺序执行代码:

library("doParallel")
foreach(i=1:3) %dopar%
  sqrt(i)

产量:

Warning message:
executing %dopar% sequentially: no parallel backend registered 

但是,如果我在启动,注册和停止集群后运行相同的代码,则会失败:

cl <- makeCluster(2)
registerDoParallel(cl)
stopCluster(cl)
rm(cl)
foreach(i=1:3) %dopar%
  sqrt(i)

产量:

Error in summary.connection(connection) : invalid connection

有没有相反的registerDoParallel()清理群集注册？还是我坚持使用旧集群的鬼魂,直到我重新开始我的R会话？

/编辑:一些谷歌搜索揭示bumphunter:::foreachCleanup()了bumphunter Biocondoctor包中的功能:

function () 
{
    if (exists(".revoDoParCluster", where = doParallel:::.options)) {
        if (!is.null(doParallel:::.options$.revoDoParCluster)) 
            stopCluster(doParallel:::.options$.revoDoParCluster)
        remove(".revoDoParCluster", envir = doParallel:::.options)
    }
}
<environment: namespace:bumphunter>

但是,此功能似乎无法解决问题.

library(bumphunter)
cl <- makeCluster(2)
registerDoParallel(cl)
stopCluster(cl)
rm(cl)
bumphunter:::foreachCleanup()
foreach(i=1:3) %dopar%
  sqrt(i)

foreach在哪里保留注册集群的信息？

我试图读取文本文件中的所有行并计划显示每个行信息.如何在循环内找到每个项目的索引？

string[] lines = File.ReadAllLines("MyFile.txt");
    List<string> list_lines = new List<string>(lines);
    Parallel.ForEach(list_lines, (line, index) =>
      {
         Console.WriteLine(index);
    //   Console.WriteLine(list_lines[index]);
         Console.WriteLine(list_lines[0]);
       });
       Console.ReadLine();

使用foreach和doMC后端使用"多核"并行性时(我使用doMC时,我调查了其他包不允许从我想要获取进度条,使用进度包,但任何进度)这适用于Linux终端,即没有tcltk弹出窗口).

鉴于它使用分叉,我可以想象它可能不可能,但我不确定.

预期用途是指示并行加载连接100个文件时的进度(通常在#!Rscript中)

我看过几个帖子,比如在R中使用"foreach()"函数时如何创建进度条？.很高兴为此奖励.

编辑

为有人向我展示500分的赏金

1. 使用foreach和多核(分叉)类型的并行性
2. 得到一个进度条
3. 使用futile.logger获取日志记录

Reprex

# load packages                                                                                                        
library("futile.logger")                                                                                               
library("data.table")                                                                                                  
library("foreach")                                                                                                     
# create temp dir                                                                                                      
tmp_dir <- tempdir()                                                                                                   
# create names for 200 files to be created                                                                             
nb_files <- 200L                                                                                                       
file_names <- file.path(tmp_dir, sprintf("file_%s.txt", 1:nb_files))                                                   
# make it reproducible                                                                                                 
set.seed(1L)                                                                                                           
nb_rows <- 1000L                                                                                                       
nb_columns <- 10L                                                                                                      
# create those 200 files sequentially                                                                                  
foreach(file_i = file_names) %do%                                                                                      
{                                                                                                                      
    DT <- as.data.table(matrix(data = runif(n = nb_rows * nb_columns), nrow …

这真让我大吃一惊.基本循环在我的计算机上花了8秒钟:

system.time({
x <- 0
for (p in 1:2) {
    for (i in 1:500) {
        for (j in 1:5000) {
            x <- x + i * j
        }
    }
}
})
x

然而,如果我foreach在非并行模式下使用,它只需要0.7秒!

system.time({
x <- 0
foreach(p = 1:2, .combine = rbind) %do% 
    for (i in 1:500) {
        for (j in 1:5000) {
            x <- x + i * j
        }
    }
})
x

结果是一样的,但foreach以某种方式能够比基本R更快地达到它!基本R的低效率在哪里？

这怎么可能？

事实上,与此相比,我获得了完全相反的结果:为什么foreach()%do%有时慢于？

我正在尝试训练几个随机森林(用于回归)让他们竞争,看看哪个特征选择和哪个参数给出最佳模型.

然而,训练似乎花了很多时间,我想知道我做错了什么.

我用于训练的数据集(train下面称为)有217k行和58列(其中只有21列作为随机森林中的预测变量.它们都是numeric或者integer,除了布尔值,它是类的character该y输出是numeric).

我跑到下面的代码四次,给值4,100,500,2000到nb_trees:

library("randomForest")
nb_trees <- #this changes with each test, see above
ptm <- proc.time()
fit <- randomForest(y ~ x1 + x2 + x3 + x4 + x5 + x6 + x7 + x8 + x9 
    + x10 + x11 + x12 + x13 + x14 + x15 + x16 + x17 + x18 + x19 …

我已成功使用以下代码在R中成功创建到TSQL服务器的并行连接:

SQL_retrieve <- function(x){
  con <-
    odbcDriverConnect(
      'driver={SQL Server};server=OPTMSLMSOFT02;database=Ad_History;trusted_connection=true'
)
  odbcGetInfo(con)
  rawData <-
    sqlQuery(con,
         paste(
           "select * from AD_MDL_R_INPUT a where a.itm_lctn_num = ",
           facility[x]
         ))
  odbcClose(con)
  return(rawData)
}

cl <- makeCluster(5)
registerDoParallel(cl)
outputPar <- foreach(j = 1:facility_count, .packages="RODBC") %dopar% SQL_retrieve(j)
stopCluster(cl)

我希望所有连接都能并行地主动下载,但实际情况是,一次只有一个或两个连接处于活动状态(见下图).这与创建的集群/连接数无关.即使有32个连接,总下载时间也会略微超过1/2(理论上应该接近1/32,对吧？).连接活动之间也有很大的暂停.为什么是这样？我在这里错过了什么？

连接利用率

请记住一些注意事项:

• TSQL服务器和R都在同一台服务器上,因此网络延迟不是问题.
• TSQL服务器允许最多约32k的连接,因此我们不会遇到会话限制问题.

更新7/26/17 再次尝试这个问题,它现在有效(代码不变).不确定从现在到最初发布之间发生了什么,但可能对MS SQL服务器设置进行了一些更改(不太可能).

拉动790万行的时间遵循下图中的曲线.

以下(简化)脚本在unix集群的主节点(4个虚拟核心)上正常工作.

library(foreach)
library(doParallel)

nc = detectCores()
cl = makeCluster(nc)
registerDoParallel(cl)

foreach(i = 1:nrow(data_frame_1), .packages = c("package_1","package_2"), .export = c("variable_1","variable_2"))  %dopar% {     

    row_temp = data_frame_1[i,]
    function(argument_1 = row_temp, argument_2 = variable_1, argument_3 = variable_2)

}

stopCluster(cl)

我想利用集群中的16个节点(16 * 4总共虚拟核心).

我想我需要做的就是更改指定的并行后端makeCluster.但是我应该怎么做呢？文档不是很清楚.

基于这个相当古老的(2013年)帖子http://www.r-bloggers.com/the-wonders-of-foreach/,似乎我应该更改默认类型(sock或者MPI- 哪个可以在unix上工作？ )

编辑

来自foreach作者的这个小插图:

默认情况下,doParallel在类Unix系统上使用多核功能,在Windows上使用snow功能.请注意,多核功能仅在一台计算机上运行任务,而不是在一组计算机上运行.但是,您可以使用snow功能在群集上执行,使用类Unix操作系统,Windows甚至组合.

什么you can use the snow functionality意思？我该怎么办？

以下代码在Windows和Ubuntu平台上产生不同的结果.据我所知,这是因为处理并行处理的方法不同.

总结:
我不能insert/ rbind数据在Linux上平行(mclapply,mcmapply),而我能做到这一点在Windows上.

感谢@Hong Ooi指出这mclapply在Windows上不起作用,但问题仍然有效.

当然,没有多个插入相同data.frame,每个插入都执行到单独的data.frame中.

library(R6)
library(parallel)

# storage objects generator
cl <- R6Class(
    classname = "cl",
    public = list(
        data = data.frame(NULL),
        initialize = function() invisible(self),
        insert = function(x) self$data <- rbind(self$data, x)
    )
)

N <- 4L # number of entities
i <- setNames(seq_len(N),paste0("n",seq_len(N)))

# random data.frames
set.seed(1)
ldt <- lapply(i, function(i) data.frame(replicate(sample(3:10,1),sample(letters,1e5,rep=TRUE))))

# entity storage
lcl1 <- lapply(i, function(i) cl$new())
lcl2 <- …

通过在使用foreach时将%dopar%更改为%do%,我可以按顺序运行代码.我该如何以编程方式执行此操作？

例如,我想要以下但只有一个foreach声明:

library(doParallel)
library(foreach)

registerDoParallel(cores = 4)

runner <- function(parallel = FALSE) {
  if (parallel)
    foreach(i=1:10) %dopar% {
      print(i)
    }
   else
    foreach(i=1:10) %do% {
      print(i)
    }
}

runner()
runner(TRUE)