R的多進(jìn)程使用與改進(jìn)

在R中需要使用多進(jìn)程時(shí)，常見(jiàn)方案是使用foreach和doParallel的組合。

foreach

foreach包中最重要的是foreach函數(shù)，該函數(shù)創(chuàng)建一個(gè)foreach對(duì)象，隨后串行或并行的執(zhí)行表達(dá)式。

library(foreach)?foreach

out:

foreach( ..., .combine, .init, .final = NULL, .inorder = TRUE, .multicombine = FALSE, .maxcombine = if (.multicombine) 100 else 2, .errorhandling = c("stop", "remove", "pass"), .packages = NULL, .export = NULL, .noexport = NULL, .verbose = FALSE)e1 %:% e2when(cond)obj %do% exobj %dopar% extimes(n)

foreach函數(shù)在創(chuàng)建時(shí)常用的幾個(gè)參數(shù)為：

...: 表達(dá)式中使用的變量。

.packages: 表達(dá)式依賴(lài)的包，字符向量。

.export: 表達(dá)式依賴(lài)的變量，字符向量。

.combine: 運(yùn)算后結(jié)果的組合方式，默認(rèn)為列表，可選?'c'?'cbind'?'rbind'?'+'?'*'等。

.errorhandling: 當(dāng)運(yùn)行中出現(xiàn)錯(cuò)誤時(shí)的處理方式。

使用時(shí)，對(duì)象后接%do%為串行，%dopar%為并行。

foreach(i=1:3) %do%????sqrt(i)

out:

[[1]][1] 1[[2]][1] 1.414214[[3]][1]?1.732051

當(dāng)使用%do%執(zhí)行時(shí)，程序?qū)?huì)自動(dòng)忽略.packages與.export變量。

如果需要使用多進(jìn)程，不只需要更換成%dopar%，你還需要注冊(cè)集群，執(zhí)行，結(jié)束集群。

library(doParallel)cl = makeCluster(4) #注冊(cè)4個(gè)進(jìn)程的集群registerDoParallel(cl)foreach(i=1:3) %dopar% sqrt(i)stopCluster(cl)?#?記得結(jié)束集群

包裝

對(duì)多進(jìn)程進(jìn)行包裝，形成runParallel函數(shù)。

library(foreach)library(doParallel)runParallel = function(FUN,PARAMS,packages = NULL,export = NULL){ cl = makeCluster(4) registerDoParallel(cl) N = length(PARAMS) R = foreach( i = 1:N, .packages = packages, .export = export, .errorhandling = 'stop' ) %dopar% { r = do.call(FUN, PARAMS[[i]]) r } stopCluster(cl) R}

程序中的do.call能夠使用提供的參數(shù)運(yùn)行FUN函數(shù)。

runParallel函數(shù)傳入FUN與并行參數(shù)的列表集合PARAMS，就可以使用FUN對(duì)每個(gè)值進(jìn)行處理，然后返回全部值。

問(wèn)題

在實(shí)際使用中遇到這樣一個(gè)問(wèn)題，在這里把問(wèn)題稍微簡(jiǎn)化一下。

有兩個(gè)文件，do_some_thing.R和do_some_other_thing.R，里面各自編寫(xiě)了一個(gè)函數(shù)。

do_some_thing.R

do_some_thing = function(x){ do_some_other_thing(x**2)}

do_some_other_thing.R

do_some_other_thing = function(x){ x / 2}

很明顯，do_some_thing.R中引用了do_some_other_thing.R中的函數(shù)。

現(xiàn)在我source這兩個(gè)文件并暴露這兩個(gè)函數(shù)，編寫(xiě)一個(gè)函數(shù)調(diào)用do_some_thing。

some_thing = new.env()source('do_some_thing.R',local = some_thing)some_other_thing = new.env()source('do_some_other_thing.R',local = some_other_thing)attach(some_thing)attach(some_other_thing)fun?=?function(x){do_some_thing(x+1)}

然后進(jìn)行多進(jìn)程調(diào)用。

params = lapply(1:10, list)runParallel(fun,params)

得到錯(cuò)誤。

Error in { : task 1 failed - "沒(méi)有"do_some_thing"這個(gè)函數(shù)"

找不到do_some_thing函數(shù)，然而當(dāng)我們加上所有可能的.export變量后。

runParallel(fun,params,export=c('do_some_thing','do_some_other_thing','some_thing','some_other_thing'))

仍然失敗。

Error in { : task 1 failed - "沒(méi)有"do_some_other_thing"這個(gè)函數(shù)"

有趣的是找不到的函數(shù)變成了do_some_other_thing，然而我明明export這個(gè)變量了啊。

在搜索中，回答普遍認(rèn)為doParallel存在設(shè)計(jì)缺陷，它不能順利的找到函數(shù)內(nèi)調(diào)用的其他自定義函數(shù)。

在不停的搜索過(guò)后，我終于找到了一個(gè)解決方案，使用doFuture包^[1]。

doFuture包會(huì)自動(dòng)解決這類(lèi)依賴(lài)問(wèn)題，無(wú)需提供.packages和.export參數(shù)，同時(shí)無(wú)需手動(dòng)結(jié)束集群。

一個(gè)更改后的版本是這樣的。

runParallel = function(FUN,PARAMS){ registerDoFuture() plan(multisession, workers = 4) #在這里指定進(jìn)程數(shù) N = length(PARAMS) R = foreach( i = 1:N, .errorhandling = 'stop' ) %dopar% { r = do.call(FUN, PARAMS[[i]]) r } R}runParallel(fun,params)

out:

[[1]][1] 2[[2]][1] 4.5······

成功。

十分推薦doFuture包！

我

我是?SSSimon Yang，關(guān)注我，用code解讀世界

References

[1]?doFuture包:?https://github.com/HenrikBengtsson/doFuture

總結(jié)

以上是生活随笔為你收集整理的找不到r低版本_R的多进程使用与改进的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。