并行运行 if 循环

Question

我有一个包含大约 400 万行的数据集，我需要对其进行循环。数据结构是重复的 ID 相互依赖，但数据在 ID 之间是独立的。对于每个 ID，第 [i+1] 行依赖于 [i]。这是一个可重现的例子。我确实意识到这个例子在内部函数方面并不实用，但它只是我所拥有的结构的一个演示。

set.seed(123)

id1 = rep(1,5)
id2 = rep(2,5)
id3 = rep(3,5)
ids = c(id1,id2,id3)

month = rep(seq(1,5),3)

x = round(rnorm(15,2,5))
y = rep(0,15)

df = as.data.frame(cbind(ids,month,x,y))

for (i in 1:nrow(df)){
  if(i>1 && df[i,1]==df[i-1,1]){
    #Main functions go here
    df[i,4] = df[i-1,4]^2+df[i,3]
  }
  else {
    df[i,4] = 1
  }
}

问题实际上是实际函数的 1000 次循环需要大约 90 秒，因此 400 万行需要几天时间。我这样跑是行不通的。然而，ID 是独立的，不需要一起运行。我的问题是:有没有办法并行运行这种类型的循环？一个非常不优雅的解决方案是将文件拆分为 50 个部分而不拆分 ID，并简单地在 50 个子文件上运行相同的代码。不过，我想应该有一种方法可以对此进行编码。

编辑:添加月份列以显示各行相互依赖的原因。解决以下两条评论:

1)实际上有6-7行函数要运行。我可以将 ifelse() 与多个函数一起使用吗？
2)所需的输出将是完整的数据帧。实际上有更多列，但我需要数据框中的每一行。

   ids month  x      y
1    1     1 -1      1
2    1     2  1      2
3    1     3 10     14
4    1     4  2    198
5    1     5  3  39207
6    2     1 11      1
7    2     2  4      5
8    2     3 -4     21
9    2     4 -1    440
10   2     5  0 193600
11   3     1  8      1
12   3     2  4      5
13   3     3  4     29
14   3     4  3    844
15   3     5 -1 712335

EDIT2: 我已经尝试应用另一篇文章中的 foreach() 包，但它似乎不起作用。这段代码会运行，但我认为问题在于行在核心之间的分布方式。如果每一行按顺序发送到不同的核心，那么相同的 ID 将永远不会在同一个核心中。

library(foreach)
library(doParallel)


set.seed(123)

id1 = rep(1,5)
id2 = rep(2,5)
id3 = rep(3,5)
ids = c(id1,id2,id3)

month = rep(seq(1,5),3)

x = round(rnorm(15,2,5))
y = rep(0,15)

df = as.data.frame(cbind(ids,month,x,y))

#setup parallel backend to use many processors
cores=detectCores()
cl <- makeCluster(cores[1]-1) #not to overload your computer
registerDoParallel(cl)

finalMatrix <- foreach(i=1:nrow(df), .combine=cbind) %dopar% {

  for (i in 1:nrow(df)){
    if(i>1 && df[i,1]==df[i-1,1]){
      #Main functions go here
      df[i,4] = df[i-1,4]^2+df[i,3]
    }
    else {
      df[i,4] = 1
    }
  }
}
#stop cluster
stopCluster(cl)

Answer 1

因此，只需使用 Rcpp 重新编写您的循环:

#include <Rcpp.h>
using namespace Rcpp;

// [[Rcpp::export]]
NumericVector fill_y(const NumericVector& x) {

  int n = x.length();
  NumericVector y(n); y[0] = 1;
  for (int i = 1; i < n; i++) {
    y[i] = pow(y[i - 1], 2) + x[i];
  }
  return y;
}

并且，要将它应用于每个组，请使用 dplyr:

df %>%
  group_by(ids) %>%
  mutate(y2 = fill_y(x))

我认为这应该足够快，因此您不需要并行性。实际上，我在@Val 的 testdat 上运行了它，只用了 2 秒(使用旧计算机)。

告诉我是否可以。否则，我会制作平行版本。