将多核与 Snow Cluster 相结合答案

【问题标题】：Combining Multicore with Snow Cluster将多核与 Snow Cluster 相结合
【发布时间】：2012-07-22 09:25:54
【问题描述】：

Parallel R 相当新。快速提问。我有一个计算密集型的算法。幸运的是，它可以很容易地分解成碎片以使用multicore 或snow。我想知道的是在实践中将multicore与snow结合使用是否被认为是好的？

我想做的是将负载拆分为在集群中的多台机器上运行，并为每台机器运行。我想利用机器上的所有内核。对于这种类型的处理，雪与multicore混合是否合理？

【问题讨论】：

（您可以使用R >=2.14.0 中包含的parallel 包，它基于multicore 和snow）。我会想到像cl <- makeCluster(...); clusterEvalQ(cl, { library(parallel); cl <- makeCluster(4); parallel_custom_function <- function(cl, ...) {...}} 这样的东西，但我很想看到一些工作代码！
你是个结合者。塞格就是一个很好的例子。绝对没有理由不能将多核与 segue 结合使用。关键是分散您的负载并了解您在哪里使用开销。
install.packages 中的警告：包‘parallel’不可用（对于 R 版本 3.0.1）

标签： r parallel-processing multicore snow

【解决方案1】：

我使用了 lockoff 上面建议的方法，即使用并行包将令人尴尬的并行工作负载分配到具有多个内核的多台机器上。首先，工作负载分布在所有机器上，然后每台机器的工作负载分布在它的所有内核上。这种方式的缺点是机器之间没有负载均衡（至少我不知道怎么做）。

所有加载的 r 代码应该是相同的，并且在所有机器 (svn) 上的相同位置。因为初始化集群需要相当长的时间，下面的代码可以通过重用创建的集群来改进。

foo <- function(workload, otherArgumentsForFoo) {
    source("/home/user/workspace/mycode.R")
    ...
}

distributedFooOnCores <- function(workload) {
    # Somehow assign a batch number to every record
    workload$ParBatchNumber = NA
    # Split the assigned workload into batches according to DistrParNumber
    batches = by(workload, workload$ParBatchNumber, function(x) x)

    # Create a cluster with workers on all machines 
    library("parallel")
    cluster = makeCluster(detectCores(), outfile="distributedFooOnCores.log")
    batches = parLapply(cluster, batches, foo, otherArgumentsForFoo)
    stopCluster(cluster)

    # Merge the resulting batches
    results = someEmptyDataframe
    p = 1;
    for(i in 1:length(batches)){
        results[p:(p + nrow(batches[[i]]) - 1), ] = batches[[i]]
        p = p + nrow(batches[[i]])      
    }

    # Clean up
    workload$ParBatchNumber = NULL
    return(invisible(results))
}

distributedFooOnMachines <- function(workload) {
    # Somehow assign a batch number to every record
    workload$DistrBatchNumber = NA
    # Split the assigned activity into batches according to DistrBatchNumber
    batches = by(workload, workload$DistrBatchNumber, function(x) x)

    # Create a cluster with workers on all machines 
    library("parallel")
    # If makeCluster hangs, please make sure passwordless ssh is configured on all machines
    cluster = makeCluster(c("machine1", "etc"), master="ub2", user="", outfile="distributedFooOnMachines.log")
    batches = parLapply(cluster, batches, foo, otherArgumentsForFoo)
    stopCluster(cluster)

    # Merge the resulting batches
    results = someEmptyDataframe
    p = 1;
    for(i in 1:length(batches)){
        results[p:(p + nrow(batches[[i]]) - 1), ] = batches[[i]]
        p = p + nrow(batches[[i]])      
    }

    # Clean up
    workload$DistrBatchNumber = NULL
    return(invisible(results))
}

我对如何改进上述方法很感兴趣。

【讨论】：