doParallel (пакет) foreach не работает для больших итераций в R

Я думаю, у тебя мало памяти. Вот модифицированная версия этого примера, которая должна работать лучше, когда у вас много задач. Он использует doSNOW, а не doParallel, потому что doSNOW позволяет вам обрабатывать результаты с помощью функции комбинирования по мере того, как они возвращаются рабочими процессами. В этом примере эти результаты записываются в файл, чтобы использовать меньше памяти, однако он считывает результаты обратно в память в конце с помощью функции «.final», но вы можете пропустить это, если у вас недостаточно памяти.

library(doSNOW)
library(tcltk)
nw <- 4  # number of workers
cl <- makeSOCKcluster(nw)
registerDoSNOW(cl)

x <- iris[which(iris[,5] != 'setosa'), c(1,5)]
niter <- 15e+6
chunksize <- 4000  # may require tuning for your machine
maxcomb <- nw + 1  # this count includes fobj argument
totaltasks <- ceiling(niter / chunksize)

comb <- function(fobj, ...) {
  for(r in list(...))
    writeBin(r, fobj)
  fobj
}

final <- function(fobj) {
  close(fobj)
  t(matrix(readBin('temp.bin', what='double', n=niter*2), nrow=2))
}

mkprogress <- function(total) {
  pb <- tkProgressBar(max=total,
                      label=sprintf('total tasks: %d', total))
  function(n, tag) {
    setTkProgressBar(pb, n,
      label=sprintf('last completed task: %d of %d', tag, total))
  }
}
opts <- list(progress=mkprogress(totaltasks))
resultFile <- file('temp.bin', open='wb')

r <-
  foreach(n=idiv(niter, chunkSize=chunksize), .combine='comb',
          .maxcombine=maxcomb, .init=resultFile, .final=final,
          .inorder=FALSE, .options.snow=opts) %dopar% {
    do.call('c', lapply(seq_len(n), function(i) {
      ind <- sample(100, 100, replace=TRUE)
      result1 <- glm(x[ind,2]~x[ind,1], family=binomial(logit))
      coefficients(result1)
    }))
  }

Я включил индикатор выполнения, так как выполнение этого примера занимает несколько часов.

Обратите внимание, что в этом примере также используется функция idiv из пакета iterators для увеличения объема работы в каждой из задач. Этот метод называется разбиением на фрагменты и часто повышает производительность параллельных вычислений. Однако использование idiv искажает индексы задач, поскольку переменная i теперь является индексом для каждой задачи, а не глобальным индексом. Для глобального индекса вы можете написать собственный итератор, который обертывает idiv:

idivix <- function(n, chunkSize) {
  i <- 1
  it <- idiv(n, chunkSize=chunkSize)
  nextEl <- function() {
    m <- nextElem(it)  # may throw 'StopIterator'
    value <- list(i=i, m=m)
    i <<- i + m
    value
  }
  obj <- list(nextElem=nextEl)
  class(obj) <- c('abstractiter', 'iter')
  obj
}

Значения, выдаваемые этим итератором, представляют собой списки, каждый из которых содержит начальный индекс и счетчик. Вот простой цикл foreach, в котором используется этот пользовательский итератор:

r <- 
  foreach(a=idivix(10, chunkSize=3), .combine='c') %dopar% {
    do.call('c', lapply(seq(a$i, length.out=a$m), function(i) {
      i
    }))
  }

Конечно, если задачи требуют достаточно интенсивных вычислений, вам может не понадобиться разбиение на фрагменты, и вы можете использовать простой цикл foreach, как в исходном примере.

Сначала я подумал, что у вас проблемы с памятью, потому что при отправке многих задач действительно используется больше памяти, и это может в конечном итоге привести к зависанию главного процесса, поэтому мой первоначальный ответ показывает несколько методов использования меньшего объема памяти. Однако теперь это звучит так, как будто есть фаза запуска и остановки, когда занят только главный процесс, но рабочие заняты в течение некоторого периода времени в середине. Я думаю, проблема в том, что задачи в этом примере на самом деле не очень требовательны к вычислительным ресурсам, поэтому, когда у вас много задач, вы начинаете действительно замечать время запуска и завершения работы. Я рассчитал время реальных вычислений и обнаружил, что каждая задача занимает всего около 3 миллисекунд. В прошлом вы не получили бы никакой выгоды от параллельных вычислений с такими маленькими задачами, но теперь, в зависимости от вашей машины, вы можете получить некоторую выгоду, но накладные расходы значительны, поэтому, когда у вас очень много задач, вы действительно заметите, что накладные расходы.

Я все еще думаю, что мой другой ответ хорошо подходит для этой проблемы, но, поскольку у вас достаточно памяти, это излишне. Наиболее важная техника использования фрагментации. Вот пример, который использует фрагментацию с минимальными изменениями по сравнению с исходным примером:

require("doParallel")
nw <- 8
registerDoParallel(nw)
x <- iris[which(iris[,5] != "setosa"), c(1,5)]
niter <- 4e+6
r <- foreach(n=idiv(niter, chunks=nw), .combine='rbind') %dopar% {
  do.call('rbind', lapply(seq_len(n), function(i) {
    ind <- sample(100, 100, replace=TRUE)
    result1 <- glm(x[ind,2]~x[ind,1], family=binomial(logit))
    coefficients(result1)
  }))
}

Обратите внимание, что это немного отличается от моего другого ответа. Он использует только одну задачу для каждого работника с использованием параметра idiv chunks, а не параметра chunkSize. Это уменьшает объем работы, выполняемой мастером, и является хорошей стратегией, если у вас достаточно памяти.