r - foreach %dopar% 写入同一个文件

Question

我想在同一个文件中并行写入，我测试了两种方法，writeLine 和 sink。

path_file <- "path"
cl <- makeCluster(4)
registerDoParallel(cl)
#fileConn<-file(path_file, "w")
sink(path_file, append=TRUE)
foreach(i = 1:3) %dopar% 
{
  #writeLines(text = paste("hellow","world", i), con = fileConn, sep="\n")
  cat(paste("hellow","world", '\n'), file = path_file, append=TRUE)
}
sink()
#close(fileConn)
parallel::stopCluster(cl)

使用 writeLine 方法我得到一个错误:{ 错误:任务 1 失败 - “不正确的连接”

使用 sink 方法，我在文件中得到了 foreach [[ ]] NULL 的结果，我不想要这个。

hellow world 1 
hellow world 2 
hellow world 3 
[[1]]
NULL

[[2]]
NULL

[[3]]
NULL

Answer 1

另一种方法是将所有输出重定向到一个文件(这可能不是你想要的)

  library(doParallel)
  library(flock)
  path_file <- "path1.txt"
  cl <- makeCluster(4,outfile=path_file)
  registerDoParallel(cl)
  foreach(i = 1:10) %dopar% 
  {
    message <- paste("hello","world", i,"\n")
    print(message)
  }
  parallel::stopCluster(cl)

或者您可能希望为每个元素创建一个文件，然后将它们连接起来

  library(doParallel)
  library(flock)
  path_file <- "path"

  cl <- makeCluster(4)
  registerDoParallel(cl)
  foreach(i = 1:103, .export ="fileConn") %dopar% 
  {
    filename = paste0(path_file,i,".txt")
    message <- paste("hello","world", i,"\n")
    print(filename)
    cat(message, file = filename, append=TRUE)
    print(message)
  }

  parallel::stopCluster(cl)

  startfile= "full.txt"
  foreach(i = 1:103, .export ="fileConn") %do% 
  {
    filename = paste0(path_file,i,".txt")
    file.append(startfile,filename)
    file.remove(filename)
  }

当多个线程试图访问同一资源时，您需要小心。为了同步对共享资源的访问，您可以使用 flock 包来设置互斥锁。(不确定为什么以下不起作用，文件连接可能不是可线程的

看看下面的代码示例

  library(doParallel)
  library(flock)
  path_file <- "path12.txt"
  fileConn<-file(path_file,open = "a")
  lock <-tempfile()

  cl <- makeCluster(4)
  registerDoParallel(cl)
  foreach(i = 1:103) %do% 
  {
    locked <- flock::lock(lock)  # Lock in order to use shared resources
    message <- paste("hello","world", i,"\n")
    cat(message, file = fileConn, append=TRUE)
    print(message)
    flock::unlock(locked)  # Release lock
    print(message)
  }

  close(fileConn)
  parallel::stopCluster(cl)