问R parLapply不平行
EN

Stack Overflow用户

提问于 2017-04-06 19:29:50

回答 1查看 1.8K关注 0票数 4

我目前正在开发一个R包，它将使用并行计算来解决一些任务，通过“并行”程序包。

当我使用包的内部函数定义的集群时，我遇到了一些非常尴尬的行为，其中parLapply函数将一个作业分配给一个工人，然后等待它完成为下一个工作分配一个任务。或者至少通过观察日志文件"cluster.log“和unix中正在运行的进程列表，这似乎是正在发生的事情。

下面是在我的包中声明的原始函数的模拟版本：

.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)

出现了描述的问题。

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

我是不是漏掉了什么？提前谢谢。

rparallel

parallel-processing

jobs

worker

人脸核身产品特惠

用户政务民生、金融、通信运营商等场景，接入全面，安全可靠新用户3折起

回答 1

Stack Overflow用户

回答已采纳

发布于 2017-08-01 13:16:53

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

我有几个改进的建议：

不要在parSolver中定义worker函数
不要将所有的varMatrix导出给每个工人
在parSolver之外创建集群

第一点很重要，因为正如您现在的例子所示，parSolver中定义的所有变量都将与匿名工作者函数一起序列化，并通过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)