我喜欢里面的环境.progress = 'text'。但是，由于列表项被发送到各个核心，然后在最后进行整理，因此不知道（从 package ）plyr's llply走了多远，这让我非常焦虑。mclapplymulticore
我一直在输出这样*currently in sim_id # ....*的消息，但这并不是很有帮助，因为它没有让我知道列表项的完整百分比（尽管知道我的脚本没有卡住并继续移动很有帮助）。
有人可以提出其他想法，让我查看我的.Rout文件并获得进步感吗？我考虑过添加一个手动计数器，但看不到我将如何实现它，因为mclapply必须在它可以给出任何反馈之前完成所有列表项的处理。
由于会mclapply产生多个进程，因此可能需要使用 fifos、管道甚至套接字。现在考虑以下示例：
library(multicore)
finalResult <- local({
f <- fifo(tempfile(), open="w+b", blocking=T)
if (inherits(fork(), "masterProcess")) {
# Child
progress <- 0.0
while (progress < 1 && !isIncomplete(f)) {
msg <- readBin(f, "double")
progress <- progress + as.numeric(msg)
cat(sprintf("Progress: %.2f%%\n", progress * 100))
}
exit()
}
numJobs <- 100
result <- mclapply(1:numJobs, function(...) {
# Dome something fancy here
# ...
# Send some progress update
writeBin(1/numJobs, f)
# Some arbitrary result
sample(1000, 1)
})
close(f)
result
})
cat("Done\n")
在这里，一个临时文件被用作先进先出，主进程分叉一个子进程，其唯一职责是报告当前进度。主进程通过调用mclapplywhere 将被评估的表达式（更准确地说，表达式块）通过writeBin.
由于这只是一个简单的示例，您可能必须根据需要调整整个输出内容。！
本质上添加了@fotNelson 解决方案的另一个版本，但进行了一些修改：
parallel而不是multicore现在已从 CRAN 中删除希望这可以帮助某人...
library(parallel)
#-------------------------------------------------------------------------------
#' Wrapper around mclapply to track progress
#'
#' Based on http://stackoverflow.com/questions/10984556
#'
#' @param X a vector (atomic or list) or an expressions vector. Other
#' objects (including classed objects) will be coerced by
#' ‘as.list’
#' @param FUN the function to be applied to
#' @param ... optional arguments to ‘FUN’
#' @param mc.preschedule see mclapply
#' @param mc.set.seed see mclapply
#' @param mc.silent see mclapply
#' @param mc.cores see mclapply
#' @param mc.cleanup see mclapply
#' @param mc.allow.recursive see mclapply
#' @param mc.progress track progress?
#' @param mc.style style of progress bar (see txtProgressBar)
#'
#' @examples
#' x <- mclapply2(1:1000, function(i, y) Sys.sleep(0.01))
#' x <- mclapply2(1:3, function(i, y) Sys.sleep(1), mc.cores=1)
#'
#' dat <- lapply(1:10, function(x) rnorm(100))
#' func <- function(x, arg1) mean(x)/arg1
#' mclapply2(dat, func, arg1=10, mc.cores=2)
#-------------------------------------------------------------------------------
mclapply2 <- function(X, FUN, ...,
mc.preschedule = TRUE, mc.set.seed = TRUE,
mc.silent = FALSE, mc.cores = getOption("mc.cores", 2L),
mc.cleanup = TRUE, mc.allow.recursive = TRUE,
mc.progress=TRUE, mc.style=3)
{
if (!is.vector(X) || is.object(X)) X <- as.list(X)
if (mc.progress) {
f <- fifo(tempfile(), open="w+b", blocking=T)
p <- parallel:::mcfork()
pb <- txtProgressBar(0, length(X), style=mc.style)
setTxtProgressBar(pb, 0)
progress <- 0
if (inherits(p, "masterProcess")) {
while (progress < length(X)) {
readBin(f, "double")
progress <- progress + 1
setTxtProgressBar(pb, progress)
}
cat("\n")
parallel:::mcexit()
}
}
tryCatch({
result <- mclapply(X, ..., function(...) {
res <- FUN(...)
if (mc.progress) writeBin(1, f)
res
},
mc.preschedule = mc.preschedule, mc.set.seed = mc.set.seed,
mc.silent = mc.silent, mc.cores = mc.cores,
mc.cleanup = mc.cleanup, mc.allow.recursive = mc.allow.recursive
)
}, finally = {
if (mc.progress) close(f)
})
result
}
该pbapply软件包已针对一般情况实现了这一点（即在类 Unix 和 Windows 上，也适用于 RStudio）。两者pblapply都有pbsapply争论cl。从文档中：
可以通过
cl参数启用并行处理。parLapply当cl是'<code>cluster'对象时mclapply被调用，当cl是整数时被调用。与没有进度条的函数的并行等效项相比，显示进度条增加了主进程和节点/子进程之间的通信开销。当进度条被禁用（即getOption("pboptions")$type == "none"dopb()isFALSE）时，这些函数会退回到它们的原始等效项。interactive()这是if时的默认设置FALSE（即从命令行 R 脚本调用）。
如果不提供cl（或通过NULL），则使用默认的非并行lapply，还包括进度条。
这是一个基于@fotNelton 解决方案的函数，适用于您通常使用 mcapply 的任何地方。
mcadply <- function(X, FUN, ...) {
# Runs multicore lapply with progress indicator and transformation to
# data.table output. Arguments mirror those passed to lapply.
#
# Args:
# X: Vector.
# FUN: Function to apply to each value of X. Note this is transformed to
# a data.frame return if necessary.
# ...: Other arguments passed to mclapply.
#
# Returns:
# data.table stack of each mclapply return value
#
# Progress bar code based on https://stackoverflow.com/a/10993589
require(multicore)
require(plyr)
require(data.table)
local({
f <- fifo(tempfile(), open="w+b", blocking=T)
if (inherits(fork(), "masterProcess")) {
# Child
progress <- 0
print.progress <- 0
while (progress < 1 && !isIncomplete(f)) {
msg <- readBin(f, "double")
progress <- progress + as.numeric(msg)
# Print every 1%
if(progress >= print.progress + 0.01) {
cat(sprintf("Progress: %.0f%%\n", progress * 100))
print.progress <- floor(progress * 100) / 100
}
}
exit()
}
newFun <- function(...) {
writeBin(1 / length(X), f)
return(as.data.frame(FUN(...)))
}
result <- as.data.table(rbind.fill(mclapply(X, newFun, ...)))
close(f)
cat("Done\n")
return(result)
})
}
您可以使用您的系统回显函数从您的工作人员中写入，因此只需将以下行添加到您的函数中：
myfun <- function(x){
if(x %% 5 == 0) system(paste("echo 'now processing:",x,"'"))
dosomething(mydata[x])
}
result <- mclapply(1:10,myfun,mc.cores=5)
> now processing: 5
> now processing: 10
如果您传递一个索引，这将起作用，例如，不是传递数据列表，而是传递索引并在工作函数中提取数据。
根据@fotNelson 的回答，使用进度条而不是逐行打印并使用 mclapply 调用外部函数。
library('utils')
library('multicore')
prog.indic <- local({ #evaluates in local environment only
f <- fifo(tempfile(), open="w+b", blocking=T) # open fifo connection
assign(x='f',value=f,envir=.GlobalEnv)
pb <- txtProgressBar(min=1, max=MC,style=3)
if (inherits(fork(), "masterProcess")) { #progress tracker
# Child
progress <- 0.0
while (progress < MC && !isIncomplete(f)){
msg <- readBin(f, "double")
progress <- progress + as.numeric(msg)
# Updating the progress bar.
setTxtProgressBar(pb,progress)
}
exit()
}
MC <- 100
result <- mclapply(1:MC, .mcfunc)
cat('\n')
assign(x='result',value=result,envir=.GlobalEnv)
close(f)
})
.mcfunc<-function(i,...){
writeBin(1, f)
return(i)
}
必须将 fifo 连接分配给 .GlobalEnv 才能从 mclapply 调用之外的函数中使用它。感谢您的问题和以前的答复，我一直想知道如何做到这一点。