furrr::future_apply内存占用与并行性能异常问题问询

问题是否正常？ #

这种内存陡增、worker数量增加反而耗时上升的情况完全不正常。对比sys.sleep(1)的正常并行表现，说明问题并非来自并行框架的基础功能，而是出在数据传递、内存管理或任务粒度的设计上。

核心原因 #

1. 数据传递开销过高：multisession模式下，主进程需将任务数据序列化后分发到worker，worker完成后还要传递状态（即使无返回值）。处理百万级大列表时，频繁的序列化/反序列化会产生巨大的CPU和内存开销，worker越多，总分发开销越大，直接抵消并行收益。
2. 内存副本累积：每个worker都会复制任务数据的副本，worker数量越多，内存中的数据副本总量越高，导致内存陡增；同时R的垃圾回收无法跨进程生效，worker进程的内存可能无法及时释放。
3. 任务粒度太细：单个正态向量求和的计算量极小，并行调度的开销远大于计算本身，worker越多，调度开销占比越高，耗时自然上升。

可行解决方案 #

1. 优化任务粒度 #

合并小任务为大批次，减少调度和数据传递次数：

library(furrr)
# 生成测试数据
big_list <- replicate(1e6, rnorm(1), simplify = FALSE)
# 拆分为1000个元素一组的批次
batch_size <- 1000
batches <- split(big_list, ceiling(seq_along(big_list)/batch_size))
# 批次处理函数
process_batch <- function(batch) {
  purrr::walk(batch, ~ sum(.x))
}
# 并行执行
plan(multisession, workers = 6)
future_walk(batches, process_batch)
plan(sequential)

2. 强化内存管理 #

• 手动触发垃圾回收，限制全局变量传递大小：

# 限制全局变量传递上限为1GB
options(future.globals.maxSize = 1024^3)
# 在批次处理末尾添加GC
process_batch <- function(batch) {
  purrr::walk(batch, ~ sum(.x))
  gc()
}

• 显式指定需传递给worker的变量，避免自动捕获冗余全局对象：

future_walk(batches, process_batch, globals = c("process_batch"))

3. 调整并行模式（适配Windows生产环境） #

• 尝试callr模式，其内存管理更高效：

library(future.callr)
plan(callr, workers = 6)
future_walk(batches, process_batch)
plan(sequential)

• 改用foreach+doParallel，部分场景下内存控制更灵活：

library(foreach)
library(doParallel)
cl <- makeCluster(6)
registerDoParallel(cl)
foreach(batch = batches) %dopar% {
  purrr::walk(batch, ~ sum(.x))
  gc()
}
stopCluster(cl)

4. 针对时序模型任务的专属优化 #

• 提前初始化worker：在worker启动时预加载模型/全局数据，避免每个任务重复加载：

plan(multisession, workers = 6)
# 给每个worker初始化加载模型
future::clusterApply(future::workers(), function() {
  # 替换为你的模型加载代码
  # model <- readRDS("your_model.rds")
})

• 直接在worker中输出CSV：确保模型推理函数直接写入文件，不返回大对象，彻底消除结果传递开销。

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