如何在foreach并行循环中赋值向量并生成DataFrame？

问题描述 #

需求是并行统计指定目录下文件的行数，将行数低于阈值（1e5）的文件移动到目标目录，最终生成包含所有文件名和对应行数的TSV文件。具体场景为：raw目录下有foo1-foo5共5个文件，foo1有100行，其余各有1e6行，用5核并行处理，每个文件分配一个工作进程。

现有代码的核心问题 #

1. 并行变量作用域错误：原代码中主进程的lines.and.names无法被子进程修改，每个子进程的lines.and.names都是局部变量，最终主进程的DataFrame仍为空。
2. wc -l输出处理错误：wc -l的输出格式是[行数] [文件名]，直接转成数值会失败，需要提取纯数字部分。
3. 文件路径缺失：未给文件名添加raw/前缀，若当前工作目录不是raw，wc和mv命令会找不到文件。
4. 目录创建不健壮：mkdir命令在目录已存在时会报错，需添加-p参数避免该问题。

修正后的代码 #

library(parallel)
library(foreach)
library(readr)

# 创建并行集群
mycluster <- makePSOCKcluster(nnodes = 5, outfile = "")
doParallel::registerDoParallel(cl = mycluster)

# 设置行数阈值
line.cutoff <- 1e5

# 获取目标目录下的文件列表（带路径前缀）
files.to.count <- list.files("raw", full.names = TRUE)
# 提取纯文件名用于后续记录
file.names <- basename(files.to.count)

# 创建目标目录（-p参数确保目录不存在时创建，已存在时不报错）
target_dir <- paste0("files-with-less-than-", line.cutoff, "-lines")
system(paste("mkdir -p", target_dir))

# 并行处理每个文件，自动合并结果到DataFrame
lines.and.names <- foreach(i = 1:length(files.to.count), .combine = rbind, .packages = "readr") %dopar% {
  file_path <- files.to.count[i]
  file_name <- file.names[i]
  
  # 统计文件行数，提取纯数字部分
  wc_output <- system(paste("wc -l", file_path), intern = TRUE)
  lines <- as.numeric(strsplit(wc_output, "\\s+")[[1]][1])
  
  # 移动符合条件的文件
  if (lines <= line.cutoff) {
    system(paste("mv", file_path, target_dir))
  }
  
  # 返回当前文件的结果行
  data.frame(name = file_name, lines.in.file = lines, stringsAsFactors = FALSE)
}

# 写入结果TSV
write_tsv(lines.and.names, "lines_and_names.txt")

# 关闭集群
stopCluster(mycluster)

关键修正点说明 #

• 使用foreach的.combine = rbind参数，让每个子进程返回的单行DataFrame自动合并成最终结果。
• 用full.names = TRUE获取带路径的文件名，确保命令能找到文件。
• 处理wc -l的输出，用正则分割提取行数的纯数字值。
• 用mkdir -p创建目标目录，避免重复创建报错。
• 移除主进程中初始化空DataFrame的冗余代码，直接接收foreach的返回值。

更简便的并行实现方案 #

方案1：使用furrr包（语法更简洁） #

furrr是purrr的并行版本，语法直观，无需手动管理集群：

library(furrr)
library(readr)

line.cutoff <- 1e5
files.to.count <- list.files("raw", full.names = TRUE)
target_dir <- paste0("files-with-less-than-", line.cutoff, "-lines")
system(paste("mkdir -p", target_dir))

# 设置并行工作数
plan(multisession, workers = 5)

# 并行处理每个文件
lines.and.names <- future_map_dfr(files.to.count, function(file_path) {
  file_name <- basename(file_path)
  wc_output <- system(paste("wc -l", file_path), intern = TRUE)
  lines <- as.numeric(strsplit(wc_output, "\\s+")[[1]][1])
  
  if (lines <= line.cutoff) {
    system(paste("mv", file_path, target_dir))
  }
  
  data.frame(name = file_name, lines.in.file = lines, stringsAsFactors = FALSE)
})

write_tsv(lines.and.names, "lines_and_names.txt")
plan(sequential) # 恢复串行执行

方案2：使用parallel::mclapply（适合Linux/macOS） #

mclapply是Unix-like系统下更轻量的并行方式，无需显式创建集群：

library(parallel)
library(readr)

line.cutoff <- 1e5
files.to.count <- list.files("raw", full.names = TRUE)
target_dir <- paste0("files-with-less-than-", line.cutoff, "-lines")
system(paste("mkdir -p", target_dir))

# 并行处理
result_list <- mclapply(files.to.count, function(file_path) {
  file_name <- basename(file_path)
  wc_output <- system(paste("wc -l", file_path), intern = TRUE)
  lines <- as.numeric(strsplit(wc_output, "\\s+")[[1]][1])
  
  if (lines <= line.cutoff) {
    system(paste("mv", file_path, target_dir))
  }
  
  data.frame(name = file_name, lines.in.file = lines, stringsAsFactors = FALSE)
}, mc.cores = 5)

# 合并结果
lines.and.names <- do.call(rbind, result_list)
write_tsv(lines.and.names, "lines_and_names.txt")

内容的提问来源于stack exchange，提问作者Alex Krohn