r - future 的解决方案

Question

我正在处理一个大型数据集，我用它来进行某些计算。由于它是一个巨大的数据集，我正在使用的机器做这项工作的时间过长，因此我决定使用 future 包来在多台机器之间分配工作并加快计算速度。 所以，我的问题是，在未来(使用 putty 和 ssh)我可以连接到那些机器(并行)，但工作本身是主要的，没有任何分配。也许你可以建议一些解决方案:

• 如何让它在所有机器上工作；
• 另外，如何检查进程是否正常工作(我的意思是某些功能或任何有助于验证这些功能的功能，如果它存在的话)。

我的代码:

library(future)
workers <- c("000.000.0.000", "111.111.1.111")
plan(remote, envir = parent.frame(), workers= workers, myip = "222.222.2.22")
start <- proc.time()
cl <- makeClusterPSOCK(
 c("000.000.0.000", "111.111.1.111"), user = "...", 
rshcmd = c("plink", "-ssh", "-pw",  "..."),  
rshopts = c("-i", "V:\\vbulavina\\privatekey.ppk"),
homogeneous = FALSE))
setwd("V:/vbulavina/r/inversion")
a <- source("fun.r")
f <- future({source("pasos.r")})
l <- future({source("pasos2.R")})
time_elapsed_parallel <- proc.time() - start
time_elapsed_parallel

f 和 l 对象应该是并行完成的，但是主机器正在做所有的工作，所以如果我能做些什么我有点困惑。

PS:我尝试了 plan() 和 remote, multiprocess, multisession, cluster 但没有。

PS2:我的本地机器是 Windows 并尝试连接到 Kubuntu 和 Debian(所有这些都关闭了防火墙)。

提前致谢。

Answer 1

作者 future这里。首先，确保您可以设置 PSOCK 集群，即通过 SSH 连接到两个 worker 并在它们上运行 Rscript。你这样做:

library(future)
workers <- c("000.000.0.000", "111.111.1.111")
cl <- makeClusterPSOCK(workers, user = "...",
                       rshcmd = c("plink", "-ssh", "-pw",  "..."),
                       rshopts = c("-i", "V:/vbulavina/privatekey.ppk"),
                       homogeneous = FALSE)
print(cl)
### socket cluster with 2 nodes on hosts '000.000.0.000', '111.111.1.111'

(如果上述 makeClusterPSOCK() 停止或不起作用，请添加参数 verbose = TRUE 以获取更多信息 - 请随时在此处报告。)

接下来，设置 PSOCK 集群后，告诉 future 的系统并行处理这两个 worker:

plan(cluster, workers = cl)

测试 futures 实际上是 resolved remotes，例如

f <- future(Sys.info()[["nodename"]])
print(value(f))
### [1] "000.000.0.000"

我暂时保留剩余的部分，它也需要调整 - 让我们确保首先让 worker 启动并运行。

继续，在并行处理中使用 source() 会使事情变得复杂，尤其是在不同机器上完成并行化时。例如，在另一台机器上调用 source("my_file.R") 要求文件 my_file.R 在该机器上也可用。即使是这样，当涉及到需要导出到外部机器的变量的自动识别时，它也会使事情复杂化。一种更安全的方法是将所有代码合并到主脚本中。说了这么多，你可以尝试替换:

f <- future({source("pasos.r")})
l <- future({source("pasos2.R")})

与

futureSource <- function(file, envir = parent.frame(), ...) {
  expr <- parse(file)
  future(expr, substitute = FALSE, envir = envir, ...)
}

f <- futureSource("pasos.r")
l <- futureSource("pasos2.R")

只要 pasos.r 和 pasos2.R 不在内部调用 source()，这个 c/应该可以工作。

顺便问一下，您使用的是什么版本的 Windows？因为使用最新的 Windows 10，您已经内置了对 SSH 的支持，您不再需要使用 PuTTY。

更新 2018-07-31:继续回答有关在 future 中使用 source() 的问题。