Проблема с doParallel и foreach. Я могу зарегистрировать ядра, но они не работают

Question

У меня большой набор данных (около 40 миллионов строк x 4 столбца), и я хочу выполнить тест Фишера для данных в каждой строке.

Пример данных выглядит так:

refAppleBase altAppleBase refHawBase altHawBase
1          115            1         94          0
2          117            4         93          1
3          125            4         94          1
4          107           26         89         12
5           87           53         66         38
6           68           58         64         32

Я написал следующий скрипт, который по существу берет каждую строку, конвертирует в матрицу, чтобы она могла работать на базовой функции fisher.test() в R, а затем выплевывает отношение шансов и p-значение.

fisher.odds.pval <- function(table){ 
fisher <- fisher.test(matrix(unlist(table), nrow=2, ncol=2))
p.val <- fisher$p.value
odds <- unname(fisher$estimate)
return(cbind(odds, p.val))
}

Теперь, очевидно, это немного неуклюже, и я хочу запустить его через 40 миллионов строк, поэтому, чтобы сэкономить время, я написал следующий скрипт, используя пакеты foreach и doParallel для распараллеливания этого между несколькими ядрами.

library(doParallel)
library(foreach)

cl <- makeCluster(10)
registerDoParallel(cl)
results <- foreach(i=1:nrow(dat)) %dopar% {
  fisher.odds.pval(table=dat[i,])
}
stopCluster(cl)

Я использовал doParallel в прошлом с большим успехом. Однако при запуске вышеуказанного скрипта я вижу, как ядра "просыпаются" и загружают данные, но затем сразу же засыпают. Тогда кажется, что только одно ядро ​​выполняет все вычисления. Вот снимок экрана top, когда работает приведенный выше код.

захват верхнего экрана

Примечание. Когда я запускаю приведенный выше скрипт для меньшего набора данных, используя %do% вместо %dopar%, он работает, поэтому я подозреваю, что между способами foreach и doParallel происходит что-то подозрительное? Но действительно потерял здесь и сейчас. Любая мысль с благодарностью.

Answer 1

Основная проблема здесь в том, что, если вы используете Windows, R должен передать dat в каждый кластер (который работает медленно и использует много памяти). Одним из возможных решений будет использование общей памяти (больше информации там ).

Воспроизводимые данные

df <- read.table(text = "refAppleBase altAppleBase refHawBase altHawBase
1          115            1         94          0
                 2          117            4         93          1
                 3          125            4         94          1
                 4          107           26         89         12
                 5           87           53         66         38
                 6           68           58         64         32")
dat <- df[rep(1:4, 1e7), ] 

fisher.odds.pval <- function(table){ 
  fisher <- fisher.test(matrix(unlist(table), nrow=2, ncol=2))
  p.val <- fisher$p.value
  odds <- unname(fisher$estimate)
  return(cbind(odds, p.val))
}

Ваше текущее решение (использует много памяти !!)

library(doParallel)
registerDoParallel(cl <- makeCluster(10))
results <- foreach(i=1:100) %dopar% {
  fisher.odds.pval(table=dat[i,])
}
stopCluster(cl)

Одно решение с использованием общей памяти

library(doParallel)
# devtools::install_github("privefl/bigstatsr")
fbm <- bigstatsr::as_FBM(dat, type = "integer")
registerDoParallel(cl <- makeCluster(10))
results2 <- foreach(i=1:100) %dopar% {
  fisher.odds.pval(table=fbm[i,])
}
stopCluster(cl)

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

Answer 2

Я думаю, что я повторяю то же поведение в Windows. Функция makeCluster() принадлежит пакету parallel и их бэкэнду для распараллеливания, который отличается от бэкэнда doParallel. Это будет работать с snow и их параллельными функциями clusterApply() и т. Д.

Если вы перейдете прямо к registerDoParallel(cl = 10) или registerDoParallel(cores = 10), он зарегистрирует бэкэнд doParallel для использования с foreach() - моя система показывает правильное распределение по всем ядрам таким образом, используя вашу функцию и данные .

Чтобы остановить рабочих, используйте registerDoSEQ(). Чтобы показать количество инициализированных рабочих, используйте getDoParWorkers()