R parLapply 不平行

Question

我目前正在开发一个 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 系统中表现出来。

函数 rasterize 和 raster 在 library(raster) 中定义，导出到集群。

对我来说奇怪的部分是，如果我在全局环境中执行与 parSolver 函数完全相同的代码，那么每件事情都会顺利进行，所有 worker 同时从事一项工作，并且任务很快就会完成。
但是，如果我执行以下操作:

library(myPackage)

varMatrix <- (...)
var1 <- (...)
result <- parSolver(varMatrix, var1)

出现描述的问题。

这似乎是一个负载平衡问题，但这并不能解释为什么它在一种情况下可以正常工作，而在另一种情况下却不能。

我在这里错过了什么吗？
提前致谢。

Answer 1

我不认为 parLapply正在按顺序运行。更有可能的是，它只是运行效率低下，使它看起来是按顺序运行的。

我有一些建议来改进它:

不要在 parSolver 里面定义工作函数

不要导出所有 varMatrix给每个 worker

在 parSolver 之外创建集群

第一点很重要，因为就您的示例而言， parSolver 中定义的所有变量都是如此。将与匿名 worker 函数一起序列化并通过 parLapply 发送给 worker .通过在任何函数之外定义辅助函数，序列化不会捕获任何不需要的变量。

第二点避免了不必要的套接字 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 ，简化代码，并使其更高效。