R語言多線程運(yùn)算操作(解決R循環(huán)慢的問題)

第R語言多線程運(yùn)算操作(解決R循環(huán)慢的問題)已經(jīng)大半年沒有更新博客了。。最近都跑去寫分析報(bào)告半年沒有R

這次記錄下關(guān)于R循環(huán)（百萬級(jí)以上）死慢死慢的問題，這個(gè)問題去年就碰到過，當(dāng)時(shí)也嘗試過多線程，butfailed......昨天試了下，終于跑通了，而且過程還挺順利

step1

先查下自己電腦幾核的，n核貌似應(yīng)該選跑n個(gè)線程，線程不是越多越好，線程個(gè)數(shù)和任務(wù)運(yùn)行時(shí)間是條開口向下的拋物線，最高點(diǎn)預(yù)計(jì)在電腦的核數(shù)上。

detectCores()檢查當(dāng)前電腦可用核數(shù)我的是4所以step2選的是4

library(parallel)

cl.cores-detectCores()

step2

多線程計(jì)算

setwd(C:\\Users\\siyuanmao\\Documents\\imdada\\0-渠道投放和新人券聯(lián)動(dòng)模型\\測算)

options(scipen=3)##取消科學(xué)計(jì)數(shù)法

channel_ad_ios_data-seq(0,50000,5000)

channel_ad_android_data-seq(0,100000,10000)

library(parallel)

func-function(n){#n=1

result_data-read.csv(發(fā)券方案.csv,stringsAsFactors=FALSE)

total_coupon_solution_data-read.csv(結(jié)果表框架.csv,stringsAsFactors=FALSE)

coupon_solution_data-subset(result_data,solution==paste(方案,n,sep=))

for(iin1:11){#i=3

coupon_solution_data$channel_ad_cost[3]-5000*(i-1)

for(jin1:11){#j=5

coupon_solution_data$channel_ad_cost[4]-10000*(j-1)

solution_mark-paste(方案,n,i,j,sep=-)

coupon_solution_data$solution-solution_mark

total_coupon_solution_data-rbind(total_coupon_solution_data,coupon_solution_data)

print(solution_mark)

return(total_coupon_solution_data)

#func(10)

system.time({

x-1:7776

cl-makeCluster(4)#初始化四核心集群

results-parLapply(cl,x,func)#lapply的并行版本

res.df-do.call(rbind,results)#整合結(jié)果

stopCluster(cl)#關(guān)閉集群

df=as.data.frame(res.df)

原來非多線程的時(shí)候，我預(yù)計(jì)要跑12個(gè)小時(shí)以上，電腦發(fā)出呼呼~~的響聲，查了下Python循環(huán)會(huì)快點(diǎn)，然后改為python版（已經(jīng)很久沒有用了，連個(gè)range都不會(huì)寫，摸索了大半天才改好，但是速度還是慢==），于是改成多線程，運(yùn)行25分鐘就出結(jié)果了~~

補(bǔ)充：R語言多線程

parallel包

install.packages(parallel)

library(parallel)

包中常用函數(shù)

detectCores()檢查當(dāng)前的可用核數(shù)

clusterExport()配置當(dāng)前環(huán)境

makeCluster()分配核數(shù)

stopCluster()關(guān)閉集群

parLapply()lapply()函數(shù)的并行版本

其實(shí)R語言本來就是一門向量化語言，如果是對于一個(gè)向量的操作，使用apply函數(shù)一族能獲得比較高的效率，相比于for循環(huán)，這種高效來自于：

用C實(shí)現(xiàn)了for循環(huán)

減少對于data.frame等數(shù)據(jù)結(jié)構(gòu)等不必要的拷貝

但是很多時(shí)候，如果想更快的話，光apply函數(shù)一族還不足夠，這時(shí)候就能用上多線程。

R語言parallel包可以幫助實(shí)現(xiàn)多線程。

parLapply的簡單代碼實(shí)戰(zhàn)

檢查當(dāng)前核數(shù)

cl.cores-detectCores()

cl.cores

[1]8

啟動(dòng)集群和關(guān)閉集群

cl-makeCluster(4)#初始化四核心集群

###并行任務(wù)

stopCluster(cl)#關(guān)閉集群

parLapply執(zhí)行多線程計(jì)算

#定義計(jì)算平方函數(shù)

square-function(x)

return(x^2)

}

#利用并行計(jì)算計(jì)算平方函數(shù)

num-c(1:3)

cl-makeCluster(4)#初始化四核心集群

results-parLapply（cl,num,square）#調(diào)用parLapply并行計(jì)算平方函數(shù)

final-do.call(c,results)#整合結(jié)果

stopCluster(cl)#關(guān)閉集群

final

[1]1,4,9

思考：在如此小的計(jì)算方式下，開4個(gè)核計(jì)算是否比開一個(gè)核要快

答案：當(dāng)然是不一定，因?yàn)樯婕暗秸{(diào)度方式等額外開銷，所以不一定快，因?yàn)檎嬲⑿衅鹱饔玫牡胤皆谟诖髷?shù)據(jù)量的計(jì)算。

時(shí)間開銷對比

兩段對比代碼

#定義計(jì)算平方函數(shù)

square-function(x)

#########

#一段冗余代碼增加執(zhí)行時(shí)間

y=2*x

if(y300)

{z=y}

else

{z=x}

##########

return(x^2)

num-c(1:10000000)

#并行計(jì)算

print(system.time({

cl-makeCluster(4)#初始化四核心集群

results-parLapply（cl,num,square）#調(diào)用parLapply并行計(jì)算平方函數(shù)

final-do.call(c,results)#整合結(jié)果

stopCluster(cl)#關(guān)閉集群

用戶系統(tǒng)流逝

7.890.2719.01

#普通計(jì)算

print(system.time({

results-lapply（num,square）

final-do.call(c,results)#整合結(jié)果

用戶系統(tǒng)流逝

29.740.0029.79

顯然在數(shù)據(jù)量比較大的時(shí)候，并行計(jì)算的時(shí)間幾乎就是于核數(shù)反比。不過，也不是多開幾個(gè)核就好，注意內(nèi)存很容易超支的，每個(gè)核都分配相應(yīng)的內(nèi)存，所以要注意內(nèi)存開銷。出現(xiàn)內(nèi)存問題的時(shí)候，需要檢查是否代碼是否合理，R語言版本（64位會(huì)比32位分配的內(nèi)存大），核分配是否合理。

上一級(jí)環(huán)境中變量的引入

R語言里邊對于環(huán)境變量有著有趣的定義，一層套一層，這里不做深入展開。

類似于在c語言函數(shù)中使用全局變量，R在執(zhí)行并行計(jì)算的時(shí)候，如果需要計(jì)算的函數(shù)出現(xiàn)在全局（上一級(jí)），那么就需要聲明引入這個(gè)變量，否則將會(huì)報(bào)錯(cuò)。

#定義計(jì)算冪函數(shù)

base=2

square-function(x)

return(x^base)

num-c(1:1000000)

#利用并行計(jì)算計(jì)算冪函數(shù)

cl-makeCluster(4)#初始化四核心集群

results-parLapply（cl,num,square）#調(diào)用parLapply并行計(jì)算平方函數(shù)

final-do.call(c,results)#整合結(jié)果

stopCluster(cl)#關(guān)閉集群

#結(jié)果報(bào)錯(cuò)

ErrorincheckForRemoteErrors(val):

4nodesproducederrors;firsterror:找不到對象base

#利用并行計(jì)算計(jì)算冪函數(shù)

cl-makeCluster(4)#初始化四核心集群

clusterExport(cl,base,envir=environment())

results-parLapply（cl,num,square）#調(diào)用parLapply并行計(jì)算平方函數(shù)

final-do.call(c,results)#整合結(jié)果

stopCluster(cl)#關(guān)閉集群

final

[1]1,4,9,16,25.......

foreach包

除了parallel包以外，還有針對并行for循環(huán)的foreach包，foreach()的使用也與parLapply()類似，兩個(gè)功能也類似，其中遇到的問題也類似。

install.packages(foreach)

library(parallel)

foreach的使用

#定義計(jì)算冪函數(shù)

square-function(x)

return(x^2)

}

非并行情況的使用：

參數(shù)中的combine就是整合結(jié)果的函數(shù)，可以是c，可以是rbind，也可以是+等

results=foreach(x=c(1:3),.combine=c)%do%square(x)

results

[1]1,4,9

并行情況的使用：

注意并行情況的時(shí)候，需要與parallel包進(jìn)行配合，引入library(doParallel)。同時(shí)%do%需要改成%dopar%。另外與parallel包不一樣的是，需要多加一句registerDoParallel(cl)來注冊核進(jìn)行使用。

cl-makeCluster(4)

registerDoParallel(cl)

results=foreach(x=c(1:100000),.combine=c)%dopar%square(x)

stopCluster(cl)

上一級(jí)環(huán)境中變量的引入

同parallel包并行計(jì)算前需要clusterExport()來引入全局變量一樣，foreach也同樣需要聲明，不同的是，foreach聲明方式直接寫在foreach()的參數(shù)export里邊。

#定義計(jì)算冪函數(shù)

base=