我目前正在开发一个R包，它将使用并行计算来解决一些任务，通过"并行"包。
在使用在我的包的函数中定义的集群时，我遇到了一些非常尴尬的行为，其中 parLapply 函数将作业分配给工作线程并等待它完成以将作业分配给下一个工作线程。 或者至少这是通过观察日志文件"cluster.log"和Unix shell中正在运行的进程列表而发生的事情。
下面是在我的包中声明的原始函数的模型版本：
.parSolver <- function( varMatrix, var1 ) {
no_cores <- detectCores()
#Rows in varMatrix
rows <- 1:nrow(varMatrix[,])
# Split rows in n parts
n <- no_cores
parts <- split(rows, cut(rows, n))
# Initiate cluster
cl <- makePSOCKcluster(no_cores, methods = FALSE, outfile = "/home/cluster.log")
clusterEvalQ(cl, library(raster))
clusterExport(cl, "varMatrix", envir=environment())
clusterExport(cl, "var1", envir=environment())
rParts <- parLapply(cl = cl, X = 1:n, fun = function(x){
part <- rasterize(varMatrix[parts[[x]],], raster(var1), .....)
print(x)
return(part)
})
do.call(merge, rParts)
}
笔记：
- 我正在使用makePSOCKcluster，因为我希望代码在Windows和Unix系统上运行，尽管这个特殊的问题只在Unix系统中表现出来。 函数栅格化
- 和栅格化在库(栅格)中定义，导出到聚类。
对我来说奇怪的是，如果我在全局环境中执行函数parSolver的完全相同的代码，那么每件事都可以顺利进行，所有工人同时从事一项工作，任务很快就会完成。 但是，如果我执行以下操作：
library(myPackage)
varMatrix <- (...)
var1 <- (...)
result <- parSolver(varMatrix, var1)
出现所描述的问题。
这似乎是一个负载平衡问题，但是这并不能解释为什么它在一种情况下工作正常，而在另一种情况下则不行。
我在这里错过了什么吗？ 提前谢谢。
我认为parLapply不是按顺序运行的。更有可能的是，它只是运行效率低下，使其看起来像是按顺序运行的。
我有一些改进建议：
- 不要在
parSolver中定义工作线程函数 - 不要将所有
varMatrix导出给每个工作人员
在 parSolver外部创建群集
第一点很重要，因为正如您的示例现在所述，parSolver中定义的所有变量将与匿名工作线程函数一起序列化，并通过parLapply发送给工作线程。通过在任何函数之外定义工作器函数，序列化不会捕获任何不需要的变量。
第二点避免了不必要的套接字 I/O，并使用更少的内存，使代码更具可扩展性。
这是一个虚假但独立的示例，与您的示例相似，展示了我的建议：
# Define worker function outside of any function to avoid
# serialization problems (such as unexpected variable capture)
workerfn <- function(mat, var1) {
library(raster)
mat * var1
}
parSolver <- function(cl, varMatrix, var1) {
parts <- splitIndices(nrow(varMatrix), length(cl))
varMatrixParts <- lapply(parts, function(i) varMatrix[i,,drop=FALSE])
rParts <- clusterApply(cl, varMatrixParts, workerfn, var1)
do.call(rbind, rParts)
}
library(parallel)
cl <- makePSOCKcluster(3)
r <- parSolver(cl, matrix(1:20, 10, 2), 2)
print(r)
请注意，这利用clusterApply函数来迭代varMatrix的行块列表，以便不需要将整个矩阵发送给每个人。它还避免了调用clusterEvalQ和clusterExport，简化了代码，并使其更有效率。