通过未来的软件包实现R中的集群作业管理

Question

我想使用集群包future (支持异步计算)来创建一个集群观察者，它可以动态地向队列添加作业/从队列中删除作业。

我想要添加到我的j观察者中的一个特定功能是将需要内存的作业分发到集群中功能更强大的机器上。但是，由于我没有使用该包的经验，我不太确定我的方法(如下所示)是否有任何陷阱。具体地说，plan的后续调用是否有任何副作用，可能会把事情搞砸？有关更多详细信息，请参阅代码中的注释。

提前感谢！

library(parallel)
library(future)

slaveIPs=c("172.16.2.10","172.16.2.21")
masterIP="172.16.2.33"
workers=makePSOCKcluster(slaveIPs,master=masterIP)

#check whether PSOCK cluster was correctly set up
unlist(clusterCall(workers,function(x) unname(Sys.info()["nodename"]))
#[1] "ip-172-16-2-10" "ip-172-16-2-21"

#now the first important part that I am not sure about
#as you can see, I only use workers[1] for the first task
#is it OK to use workers[1] like that?
plan(cluster,workers=workers[1])

f=future({
  #do memory-hungry work
  unname(Sys.info()["nodename"])
})

message(value(f))
#ip-172-16-2-10

#now I am only using workers[2] for the second task
#Is this ok? Does the previous call to 'plan' need some cleaning before?
plan(cluster,workers=workers[2])

f=future({
  #do low-memory work
  unname(Sys.info()["nodename"])
})

message(value(f))
#ip-172-16-2-21

stopCluster(workers)

Answer 1

future的作者在这里：

是的，可以像这样改变未来的策略，也就是使用plan()。另一种方法是使用：

f <- cluster({
  #do low-memory work
  unname(Sys.info()["nodename"])
}, workers = workers[2])

这基本上就是内部正在发生的事情。

像这样显式指定未来策略的缺点是，您的代码将被硬编码为使用cluster未来。

仅供参考，我计划在未来添加某种机制来指定首选或所需的“资源”。这只是目前的概念，短期内不会存在，但我正在考虑与以下内容一致的东西：

f <- future({ ... }, needs = "himem")

其中可以向工作者查询himem标签/属性，例如attr(workers[2], "provides") <- c("himem", "superfast")。我分享这些想法只是为了让你知道我意识到了你的需求。同样，在这种机制可用之前，还需要相当长的一段时间，因此，在此期间，您需要明确指定上述未来的策略。

顺便说一下，而不是：

slaveIPs=c("172.16.2.10","172.16.2.21")
masterIP="172.16.2.33"
workers=makePSOCKcluster(slaveIPs,master=masterIP)

您可以尝试：

slaveIPs <- c("172.16.2.10", "172.16.2.21")
workers <- makeClusterPSOCK(slaveIPs)

由future包提供-这避免了必须知道/指定主服务器的IP地址。