r - 如何使用 foreach 克服内存限制

Question

我正在尝试处理 > 10000 个保存在磁盘上的 xts 对象，每个对象在加载到 R 时大约为 0.2 GB。我想使用 foreach 来并行处理这些对象。我的代码适用于我在内存中预加载、导出等的 100 个 xts 对象，但在超过 100 个 xts 对象后，我在我的机器上达到了内存限制。

我正在尝试做的示例:

require(TTR)
require(doMPI)
require(foreach)

test.data <- runif(n=250*10*60*24)

xts.1 <- xts(test.data, order.by=as.Date(1:length(test.data)))
xts.1 <- cbind(xts.1, xts.1, xts.1, xts.1, xts.1, xts.1)

colnames(xts.1) <- c("Open", "High", "Low", "Close", "Volume", "Adjusted")

print(object.size(xts.1), units="Gb")

xts.2 <- xts.1
xts.3 <- xts.1
xts.4 <- xts.1

save(xts.1, file="xts.1.rda")
save(xts.2, file="xts.2.rda")
save(xts.3, file="xts.3.rda")
save(xts.4, file="xts.4.rda")

names <- c("xts.1", "xts.2", "xts.3", "xts.4")

rm(xts.1)
rm(xts.2)
rm(xts.3)
rm(xts.4)

cl <- startMPIcluster(count=2) # Use 2 cores
registerDoMPI(cl)

result <- foreach(name=names, 
                  .combine=cbind, 
                  .multicombine=TRUE, 
                  .inorder=FALSE, 
                  .packages=c("TTR")) %dopar% {
    # TODO: Move following line out of worker. One (or 5, 10,
    # 20, ... but not all) object at a time should be loaded 
    # by master and exported to worker "just in time"
    load(file=paste0(name, ".rda"))

    return(last(SMA(get(name)[, 1], 10)))
}

closeCluster(cl)

print(result)

所以我想知道如何在发送/需要之前从磁盘“及时”加载每个(或几个，如 5、10、20、100……但不是全部)xts 对象/导出给 worker 。我无法在工作人员中加载对象(基于名称和存储在磁盘上的文件夹)，因为工作人员可以在远程计算机上而无需访问对象存储在磁盘上的文件夹。所以我需要能够在主进程中“及时”读取/加载它们......

我使用 doMPI 和 doRedis 作为并行后端。 doMPI 似乎内存效率更高，但比 doRedis 慢(在 100 个对象上)。

所以我想了解什么是解决这个问题的正确“策略”/“模式”。

Answer 1

除了使用 doMPI 或 doRedis，您还需要编写一个返回适当迭代器的函数。我的迭代器包中的小插图“编写自定义迭代器”中有许多示例应该会有所帮助，但这里有一个快速尝试这样的功能:

ixts <- function(xtsnames) {
  it <- iter(xtsnames)

  nextEl <- function() {
    xtsname <- nextElem(it)  # throws "StopIteration"
    load(file=paste0(xtsname, ".rda"))
    get(xtsname)
  }

  obj <- list(nextElem=nextEl)
  class(obj) <- c('ixts', 'abstractiter', 'iter')
  obj
}

这真的很简单，因为它基本上是围绕“names”变量的迭代器的包装器。小插图在几个示例中使用了这种技术。

您可以将“ixts”与 foreach 结合使用，如下所示:

result <- foreach(xts=ixts(names),
                  .combine=cbind, 
                  .multicombine=TRUE, 
                  .inorder=FALSE, 
                  .packages=c("TTR")) %dopar% {
    last(SMA(xts[, 1], 10))
}

尽管此迭代器可用于任何 foreach 后端，但并非所有后端都会及时调用它。 doMPI 和 doRedis 会，但是 doParallel 和 doMC 预先从迭代器获取所有值，因为 clusterApplyLB 和 mclapply 要求这些值都在列表中。 doMPI 和 doRedis 旨在与迭代器一起使用，以提高内存效率。