doParallel 和 foreach 无法并行化合并操作

Question

我正在尝试将一个大的 data.frame 与一个小的 data.frame 合并，并并行化计算。下面的代码完美运行，最大限度地利用了我机器的所有内核：

len <- 2000000
set.seed(666)
dat = paste(sample(letters, len, rep = T), sample(0:9, len, rep = T), sample(letters, len, rep = T), sep = '') # create a vector of strings that are 3-long
head(dat)
set.seed(777)
num <- sample(0:9, len, replace = T)
bigDF <-  data.frame(dat = dat, num = num)
smallDF <- data.frame(num = 0:9, caps = toupper(letters[1:10]))
startP <- 1
chunk <- 10000
nodes <- detectCores()
cl <- makeCluster(nodes)
registerDoParallel(cl)
mergedList <- foreach(i = 0:(len/chunk - 1)) %dopar% {
    tmpDF = bigDF[(startP + i * chunk):(startP - 1 + (i + 1) * chunk), ]
    merge(tmpDF, smallDF, by = 'num', all.x = T)
}
stopCluster(cl)

一旦我将向量 dat 更改为包含 5 长的字符串，并行性就会崩溃，尽管没有错误或警告，但只有 1 个内核参与计算：

len <- 2000000
set.seed(666)
dat = paste(sample(letters, len, rep = T), sample(0:9, len, rep = T), sample(letters, len, rep = T), sample(letters, len, rep = T), sample(letters, len, rep = T), sample(letters, len, rep = T), sep = '') # create a vector of strings that are 6-long
head(dat)
set.seed(777)
num <- sample(0:9, len, replace = T)
bigDF <-  data.frame(dat = dat, num = num)
smallDF <- data.frame(num = 0:9, caps = toupper(letters[1:10]))
startP <- 1
chunk <- 10000
nodes <- detectCores()
cl <- makeCluster(nodes)
registerDoParallel(cl)
mergedList <- foreach(i = 0:(len/chunk - 1)) %dopar% {
    tmpDF = bigDF[(startP + i * chunk):(startP - 1 + (i + 1) * chunk), ]
    merge(tmpDF, smallDF, by = 'num', all.x = T)
}
stopCluster(cl)

为什么会出现这种不一致，如何解决？在特定示例中，如果将dat 索引为整数，则代码有效。但索引并不是所有情况下的答案。为什么字符串的长度对所使用的核心数量很重要？

Answer 1

我相信不同之处在于，在第一种情况下，“bigDF”的第一列是一个有 6,760 个级别的因子，而在第二种情况下，它有 1,983,234 个级别。拥有大量级别会导致许多性能问题。当我使用stringsAsFactors=FALSE 创建“bigDF”时，性能要好得多。

bigDF <- data.frame(dat=dat, num=num, stringsAsFactors=FALSE)

我还使用了 itertools 包中的“isplitRows”函数来避免将所有“bigDF”发送给每个工人：

library(itertools)
mergedList <- foreach(splitDF=isplitRows(bigDF, chunkSize=chunk)) %dopar% {
    merge(splitDF, smallDF, by = 'num', all.x = T)
}

在我运行 R 3.1.1 的 6 核 Linux 机器上，您的第二个示例运行了大约 332 秒。当我使用stringsAsFactors=FALSE 时，它运行了大约 50 秒。当我还使用 isplitRows 时，时间下降到 5.5 秒，比您的第二个示例快了大约 60 倍。

Answer 2

还没有答案，但是： 如果我运行您的代码但使用%do% 以免并行化，则除了dat 名称之外，这两种情况我会得到相同（成功）的结果。如果我使用 %dopar% 运行短名称和使用 %do% 运行长名称，则相同。

这开始看起来像是其中一个支持包中的一个微妙错误，因此您可能希望在这个包上 ping 开发人员。

9 月 29 日更新：我运行了我认为相同的设置，但使用了 ClusterMap：

dffunc <-function(i=i,bigDF=bigDF,smallDF=smallDF,startP=startP,chunk=chunk) {
tmpDF <- bigDF[(startP + i * chunk):(startP - 1 + (i + 1) * chunk), ]
    merge(tmpDF, smallDF, by = 'num', all.x = T)
    }


clusmerge<- clusterMap(cl,  function(i) {dffunc(i=i)}, 0:(len/chunk-1),MoreArgs=list(bigDF=bigDF,smallDF=smallDF,startP=startP,chunk=chunk) )

在这种情况下，无论dat 名称字符串的长度如何，我都会启动并运行所有节点。我又开始怀疑%dopar% 或foreach 包中的其他地方存在一些错误。

作为旁注，我可以建议不要这样做

nodes <- detectCores()
cl <- makeCluster(nodes)

因为这可能会挂起您的整个机器。更好cl <- makeCluster(nodes-1) :-)