有没有办法在 xts 中将 period.apply 与 doParallel 和 foreach 一起使用？

Question

我想在 R 中并行化 period.apply 函数，我正在尝试使用 doParallel 和 Foreach，但我不知道如何实现这个函数。我使用的数据是带有日期时间索引和变量值的xts 对象，我想做的是每 5 秒取一次数据的平均值：

                                     VAR
2018-01-01 00:00:00                1945.054
2018-01-01 00:00:02                1944.940
2018-01-01 00:00:05                1945.061
2018-01-01 00:00:07                1945.255
2018-01-01 00:00:10                1945.007
2018-01-01 00:00:12                1944.995

这是我编写的代码示例，但它不起作用：

library(xts)
library(doParallel)
library(foreach)

cores <- detectCores()
cluster <- makeCluster(cores, type = "PSOCK")
registerDoParallel(cluster)

ends <- endpoints(x,"secs",5)
m <- foreach(i = 1:length(index(x))) %dopar% period.apply(x,ends,mean)
index(m) <- foreach(m) %dopar% trunc(index(m),"secs")
stopCluster()

有效的代码是这样的，但对于更大的数据库，它需要太多时间：

ends <- endpoints(x,"secs",5)
m <- period.apply(x, ends, mean)
index(m) <- trunc(index(m),"secs")

有没有更有效的方法？

提前致谢。

Answer 1

我对这个问题中说明的period.apply() 的表现感到非常沮丧。我的抑郁症变成了一种让抑郁症变得更快的痴迷。所以我用 C 重写了它。这是一个使用它并显示性能改进的示例。

library(xts)  # need the GitHub development version
period_apply <- xts:::period_apply  # not exported

set.seed(21)
x <- .xts(rnorm(1e7), 1:1e7)
e <- endpoints(x, "seconds", 5)

system.time(y <- period.apply(x, e, sum))  # current version
#    user  system elapsed 
#  77.904   0.368  78.462 
system.time(z <- period_apply(x, e, sum))  # new C version
#    user  system elapsed 
#  15.468   0.232  15.741
all.equal(y, z)
# [1] TRUE

因此，对于本示例，它的速度大约快了 5 倍。还有一些事情可以让它更快，但 5x 是一个停下来展示它可能会更好的好地方。如果您想（并且足够勇敢）尝试一下，请查看latest development version。

Answer 2

您是否在一些简单的数据集上尝试过您的代码？因为一旦我让它运行，它就会多次完成所有工作（x 中的每一行一次）。此外，如果您尝试并行化工作，通常最好让“工作人员”在发回数据之前尽可能多地完成工作。在您的代码中，您有两个连续的 foreach 调用，这会导致额外的通信开销。

我的做法是这样的：

1. 将xts 对象拆分为N 垃圾，确保我们以5 秒的间隔之一进行拆分。
2. 让每个工人为一块做所有的工作。
3. 合并结果。如何选择N？

由于split.xts 用于第一步，每个块将具有相同数量的 5s 间隔。但是，要完成的工作量（可能）更多地取决于数据点的数量，而不是 5s 间隔的数量。因此，如果这些块之间的点分布不均匀，那么使用更多的块以及一些负载平衡可能是有意义的。如果点的分布是均匀的，则将N 尽可能大以最小化通信开销是有意义的。这里我采用后一种方法，即设置N 等于内核数。

现在让我们生成一些示例数据并应用您的工作解决方案：

library(xts)
x <- xts(x = runif(100),
         order.by = as.POSIXct("2018-01-01") + 0:99)

ends <- endpoints(x,"secs",5)
m <- period.apply(x, ends, mean)
index(m) <- trunc(index(m),"secs")

接下来我们设置并行集群：

library(doParallel)
library(foreach)

cores <- detectCores()
cluster <- makeCluster(cores, type = "PSOCK")
registerDoParallel(cluster)

现在我们必须拆分xts 对象。在这里，我首先确定整个对象的时间跨度，并以N 5s 为间隔进行分配。

N <- cores
k <- as.integer(ceiling(difftime(max(index(x)), min(index(x)), units = "secs") / (5 * N)))

接下来，我将xts 对象拆分为xts 对象列表，每个对象的长度大致相同：

split_x <- split(x, f = "secs", k = 5 * k)

现在我让foreach 遍历这些块并合并结果：

m2 <- foreach(x = split_x, .packages = c("xts"), .combine = c) %dopar% {
    ends <- endpoints(x,"secs",5)
    m <- period.apply(x, ends, mean)
    index(m) <- trunc(index(m),"secs")
    m
}
stopCluster(cluster)

万岁，结果是一样的：

all.equal(m, m2)
#> [1] TRUE