R語言并行計算的原理和案例

眾所周知，在大數(shù)據(jù)時代R語言有兩個弱項，其中一個就是只能使用單線程計算。但是R在2.14版本之后，R就內(nèi)置了parallel包，強化了R的并行計算能力。

parallel包實際上整合了之前已經(jīng)比較成熟的snow包和multicore包，multicore無法在windows下運行。parallel包可以很容易的在計算集群上實施并行計算，在多個CPU核心的單機上，也能發(fā)揮并行計算的功能。我們今天就來探索一下parallel包在多核心單機上的使用。

parallel包的思路和lapply函數(shù)很相似，都是將輸入數(shù)據(jù)分割、計算、整合結果。只不過并行計算是用到了不同的cpu來運算。

這樣的計算過程可以使用如下方式來表述：

1、啟動M個附屬進程，并初始化

2、針對于任務，為每個附屬進程分發(fā)所有的數(shù)據(jù)

3、將任務粗略的分為M個塊兒（chunks），并將這些塊兒發(fā)送到附屬進程（包含需要的R代碼）

4、等待所有的附屬進程完成計算任務，并返回結果

5、對于其他任務也同樣重復2-4

6、關閉附屬進程

在parallel包里，對應上述兩種并行化方式有如下兩個核心函數(shù)（針對于lapply函數(shù)的并行化，mclapply在windows上不能使用）：

parLapply(cl, x, FUN, ...)

mclapply(X, FUN, ..., mc.cores)

案例1、不使用并行計算，直接使用lapply（隱式循環(huán)函數(shù)，它實際就是對不同的數(shù)據(jù)應用了相同的函數(shù)）：

fun <- function(x){

return (x+1);

}

system.time({

res <- lapply(1:5000000, fun);

});

user  system elapsed

21.42    1.74   25.70

案例2、使用parallel包來加速

library(parallel)

#打開四核，具體核數(shù)根據(jù)機器的核數(shù)決定

cl <- makeCluster(getOption("cl.cores", 4));

system.time({

res <- parLapply(cl, 1:5000000,  fun)

});

user system elapsed

6.54 0.34 19.95

#關閉并行計算

stopCluster(cl);

看看單核機器跑出來的結果：

user  system elapsed

29.30    9.23   97.22

所以，并非核數(shù)越多越好，看機器配置。

這個函數(shù)有兩點要注意：

首先要先用detectCores函數(shù)確定系統(tǒng)核心數(shù)目，對于Window系統(tǒng)下的Intel I5或I7 處理器，一般使用detectCores(logical = F)來獲得實際的物理核心數(shù)量。

由于這個函數(shù)使用的是調(diào)用Rscript的方式，這個例子里，對象被復制了三份，因此內(nèi)存會吃的很厲害，在大數(shù)據(jù)條件就要小心使用。

案例3、在Linux下使用mclapply函數(shù)的效果如下：

mc <- getOption("mc.cores", 3)

system.time({

res <- mclapply(1:5000000, fun, mc.cores = mc);

});

user system elapsed

6.657 0.500 7.181

foreach包是revolutionanalytics公司貢獻給R開源社區(qū)的一個包，它能使R中的并行計算更為方便。與sapply函數(shù)類似，foreach函數(shù)中的第一個參數(shù)是輸入?yún)?shù)，%do%后面的對象表示運算函數(shù)，而.combine則表示運算結果的整合方式。 下面的例子即是用foreach來完成前面的同一個任務。如果要啟用并行，則需要加載doParallel包，并將%do%改為%dopar%。這樣一行代碼就能方便的完成并行計算了。

案例4、foreach包的使用：

library(foreach)

# 非并行計算方式，類似于sapply函數(shù)的功能

x <- foreach(x=1:1000,.combine='rbind') %do% func(x)

# 啟用parallel作為foreach并行計算的后端

library(doParallel)

cl <- makeCluster(4)

registerDoParallel(cl)

# 并行計算方式

x <- foreach(x=1:1000,.combine='rbind') %dopar% func(x)

stopCluster(cl)