foreach包简介

原文地址：http://blog.163.com/yugao1986@126/blog/static/6922850820145303931857/

FROM:《Using The foreach Package》

foreach包提供了一种新的循环运行R脚本的循环结构，它支持并行运算。

#1 简介

require(foreach)

## Loading required package: foreach

#利用foreach重复运行sqrt函数
foreach(i=1:3) %do% sqrt(i)

## [[1]]
## [1] 1
## 
## [[2]]
## [1] 1.414
## 
## [[3]]
## [1] 1.732

foreach返回的是list格式值，list格式是默认的数据格式。

上例中，我们将循环值赋予i,进而将i作为函数sqrt的参数。类似，我们可以指定多个循环：

foreach(a=1:3, b=rep(10, 3)) %do% {
  a + b
}

## [[1]]
## [1] 11
## 
## [[2]]
## [1] 12
## 
## [[3]]
## [1] 13

上面a和b被赋予同样数量的值，如果只赋予两个值给b，而给a赋1000个值，我们得到的值同样只有2个：

foreach(a=1:1000, b=rep(10, 2)) %do% (a + b)

## [[1]]
## [1] 11
## 
## [[2]]
## [1] 12

#2 “.combine”选项

foreach返回的数据格式是list格式，有时我们需要向量格式的数据，此时可以使用.combine选项：

foreach(i=1:3, .combine="c") %do% exp(i)

## [1]  2.718  7.389 20.086

这里.combine选项连接了“c”函数，该函数的功能是连接所有返回值组成向量。此外，我们可以使用“cbind”将生成的多个向量组合成矩阵,例如生成四组随机数向量，进而按列合并成矩阵：

foreach(i=1:4, .combine="cbind") %do% rnorm(4)

##      result.1 result.2 result.3 result.4
## [1,]  0.26634 -0.73193 -0.25927   0.8632
## [2,]  0.54132  0.08586  1.46398  -0.6995
## [3,] -0.15619  0.85427 -0.47997   0.2160
## [4,]  0.02697 -1.40507 -0.06972   0.2252

我们还可以使用“+”或“*”等运算符作为.combine选项的值：

foreach(i=1:4, .combine="+") %do% rnorm(4)

## [1]  1.203  1.354  1.793 -2.245

也可以使用自定义函数：

cfun <- function(a, b) NULL
foreach(i=1:4, .combine="cfun") %do% rnorm(4)

## NULL

如果自定义函数涉及多个参数，我们需要使用.multicombine

cfun <- function(...) NULL
foreach(i=1:4, .combine="cfun", .multicombine=TRUE) %do% rnorm(4)

## NULL

.maxcombine可以设置最大参数个数的上线

cfun <- function(...) NULL
foreach(i=1:4, .combine='cfun', .multicombine=TRUE, .maxcombine=10) %do% rnorm(4)

## NULL

在进行并行计算时候，.inorder选项将非常重要。该参数是逻辑型的，默认为TRUE,某些情况下可以改为FALSE。

foreach(i=4:1, .combine="c") %dopar% {
  Sys.sleep(3*i)
  i
}

## Warning: executing %dopar% sequentially: no parallel backend registered

## [1] 4 3 2 1

foreach(i=4:1, .combine='c', .inorder=FALSE) %dopar% {
  Sys.sleep(3 * i)
  i
}

## [1] 4 3 2 1

#3 迭代器

迭代循环值不是必须设定位向量或者list格式，我们可以结合"iterators"包使用迭代器设置其格式。iterators包中的irnorm函数在每次调用时都可以返回指定数量的随机数：

require(iterators)

## Loading required package: iterators

foreach(a=irnorm(4,count=4), .combine='cbind') %do% a

##      result.1 result.2 result.3 result.4
## [1,]  -0.8766  -1.5256   1.1133   0.2166
## [2,]   0.1950   0.7027   0.4458   0.2943
## [3,]  -1.0098  -0.5317  -0.6918  -0.3026
## [4,]   1.0034  -0.1961  -0.6325   0.9973

在处理大数据时候，该方法相当有用，因为iterators按照我们的命令动态生成随机数，而不是在一开始就生成随机数再进行运算操作。例如，计算数千个随机数向量的汇总：

set.seed(123)
system.time(foreach(a=irnorm(4, count=1000), .combine='+') %do% a)

##    user  system elapsed 
##    1.33    0.00    1.34

#使用icount函数生成1000个随机数
set.seed(123)
system.time(foreach(icount(1000), .combine='+') %do% rnorm(4))

##    user  system elapsed 
##    1.08    0.00    1.08

#4 并行计算

foreach包创作是为了解决一些并行计算问题，将"%do%“更改为“%dopar%”前面例子就可以实现并行计算。并行计算一些小任务会比按顺序运算它们花费更多的时间，所以当普通运算足够快的时候，并没有必要使用并行计算模式改进其运算效率。

##4.1并行随机森林

下面使用随机森林来演示并行计算

#生成矩阵x作为输入值，y作为目标因子
x <- matrix(runif(500), 100)
y <- gl(2, 50)
#导入randomForest包
require(randomForest)

## Loading required package: randomForest
## randomForest 4.6-7
## Type rfNews() to see new features/changes/bug fixes.

如果我们要创建一个包含1200棵树的随机森林模型，在6核CPU电脑上，我们可以将其分割为六块执行randomForest函数六次，同时将ntree参赛设为200,最后再将结果合并。

a.按顺序执行代码

rf <- foreach(ntree=rep(200, 6), .combine=combine) %do%
  randomForest(x, y, ntree=ntree)
rf

## 
## Call:
##  randomForest(x = x, y = y, ntree = ntree) 
##                Type of random forest: classification
##                      Number of trees: 1200
## No. of variables tried at each split: 2

b.并行计算

将%do%改为“%dopar%”,同时使用.packages调用randomForest：

rf <- foreach(ntree=rep(200,6), .combine=combine, .packages="randomForest") %dopar% 
  randomForest(x, y, ntree=ntree)
rf

## 
## Call:
##  randomForest(x = x, y = y, ntree = ntree) 
##                Type of random forest: classification
##                      Number of trees: 1200
## No. of variables tried at each split: 2

##4.2并行应用apply函数

apply函数式R软件中的基本函数，下面我们看看如何实现apply的并行计算。

applyKernel <- function(newX, FUN, d2, d.call, dn.call=NULL, ...) {
  ans <- vector("list", d2)
  for(i in 1:d2) {
    tmp <- FUN(array(newX[,i], d.call, dn.call), ...)
    if(!is.null(tmp)) ans[[i]] <- tmp
    }
  ans
}
applyKernel(matrix(1:16, 4), mean, 4, 4)

## [[1]]
## [1] 2.5
## 
## [[2]]
## [1] 6.5
## 
## [[3]]
## [1] 10.5
## 
## [[4]]
## [1] 14.5

使用foreach包运行上述函数

applyKernel <- function(newX, FUN, d2, d.call, dn.call=NULL, ...){
  foreach(i=1:d2) %dopar%
     FUN(array(newX[, i], d.call, dn.call), ...)
}
applyKernel(matrix(1:16, 4), mean, 4, 4)

## [[1]]
## [1] 2.5
## 
## [[2]]
## [1] 6.5
## 
## [[3]]
## [1] 10.5
## 
## [[4]]
## [1] 14.5

该方法将导致newX数组被发送至每一个并行运算器中，有时每一个并行运算只需要newX的一列数据，此时我们就需要避免这种情况。利用iterators包迭代器遍历矩阵的列，这种方法可以解决该问题。

applyKernel <- function(newX, FUN, d2, d.call, dn.call=NULL, ...) {
  foreach(x=iter(newX, by='col')) %dopar%
    FUN(array(x, d.call, dn.call), ...)
}
applyKernel(matrix(1:16, 4), mean, 4, 4)

## [[1]]
## [1] 2.5
## 
## [[2]]
## [1] 6.5
## 
## [[3]]
## [1] 10.5
## 
## [[4]]
## [1] 14.5

上述代码只是将矩阵中的指导列赋予并行计算器，将newX矩阵放在更大的矩阵块中会更有效率。iblkcol函数可以返回一个迭代器，该迭代器返回多个列的原始矩阵。该种方法意味着对原始矩阵中的子矩阵的列运算自定义的函数：

iblkcol <- function(a, chunks) {
  n <- ncol(a)
  i <- 1
 
  nextEl <- function() {
    if (chunks <= 0 || n <= 0) stop('StopIteration')
    m <- ceiling(n / chunks)
    r <- seq(i, length=m)
    i <<- i + m
    n <<- n - m
    chunks <<- chunks - 1
    a[,r, drop=FALSE]
    }
 
  obj <- list(nextElem=nextEl)
  class(obj) <- c('abstractiter', 'iter')
  obj
  }
 
 
applyKernel <- function(newX, FUN, d2, d.call, dn.call=NULL, ...) {
  foreach(x=iblkcol(newX, 3), .combine='c', .packages='foreach') %dopar% {
    foreach(i=1:ncol(x)) %do% 
      FUN(array(x[,i], d.call, dn.call), ...)
    }
  }
applyKernel(matrix(1:16, 4), mean, 4, 4)

## [[1]]
## [1] 2.5
## 
## [[2]]
## [1] 6.5
## 
## [[3]]
## [1] 10.5
## 
## [[4]]
## [1] 14.5

#5 列表解析

foreach提供了when函数实现条件运算

foreach(a=irnorm(1, count=10), .combine='c') %:% when(a >= 0) %do% sqrt(a)

## [1] 1.079 1.185 1.465 1.396

下面是foreach结合when函数完成的简单快速排序函数示例：

qsort <- function(x) {
  n <- length(x)
  if (n == 0) {
    x
    } else {
      p <- sample(n, 1)
      smaller <- foreach(y=x[-p], .combine=c) %:% when(y <= x[p]) %do% y
      larger <- foreach(y=x[-p], .combine=c) %:% when(y > x[p]) %do% y
      c(qsort(smaller), x[p], qsort(larger))
      }
}
 
qsort(runif(12))

##  [1] 0.1481 0.2000 0.2769 0.4729 0.4747 0.5730 0.6394 0.6524 0.8315 0.8325
## [11] 0.8413 0.8724

#6 总结

大多数并行计算都主要完成三件事情：将问题分割小块、对小块问题进行并行计算、合并计算结果。foreach包中，迭代器完成分割工作，”%dopar%“函数实现对小块的并行计算，”.combine"函数完成合并工作。