R的多進程使用與改進

在R中需要使用多進程時，常見方案是使用foreach和doParallel的組合。

foreach

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

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ù)為：

...: 表達式中使用的變量。

.packages: 表達式依賴的包，字符向量。

.export: 表達式依賴的變量，字符向量。

.combine: 運算后結果的組合方式，默認為列表，可選?'c'?'cbind'?'rbind'?'+'?'*'等。

.errorhandling: 當運行中出現(xiàn)錯誤時的處理方式。

使用時，對象后接%do%為串行，%dopar%為并行。

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

out:

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

當使用%do%執(zhí)行時，程序將會自動忽略.packages與.export變量。

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

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

包裝

對多進程進行包裝，形成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ù)運行FUN函數(shù)。

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

問題

在實際使用中遇到這樣一個問題，在這里把問題稍微簡化一下。

有兩個文件，do_some_thing.R和do_some_other_thing.R，里面各自編寫了一個函數(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這兩個文件并暴露這兩個函數(shù)，編寫一個函數(shù)調用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)}

然后進行多進程調用。

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

得到錯誤。

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

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

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

仍然失敗。

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

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

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

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

doFuture包會自動解決這類依賴問題，無需提供.packages和.export參數(shù)，同時無需手動結束集群。

一個更改后的版本是這樣的。

runParallel = function(FUN,PARAMS){ registerDoFuture() plan(multisession, workers = 4) #在這里指定進程數(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，關注我，用code解讀世界

References

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

總結

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

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