R中并行处理包实现LPA模型批量拟合与多结果存储的问题咨询

问题根源 #

你遇到的问题核心来自两个底层逻辑限制：

1. 并行进程内存不共享：foreach启动的worker进程拥有独立的内存空间，无法直接修改主进程中预定义的results矩阵，你在循环内部对results[sim_num, ]的赋值仅在当前worker的局部环境生效，不会同步到主进程。
2. 默认返回规则：foreach默认返回每个迭代块中最后一行代码的求值结果，你现在的迭代块最后一行是2类模型的p值赋值，所以最终只能拿到零散的单个值。

同时你当前的代码还存在隐性问题：所有并行迭代都共用df.dat、lpa1_15.inp等同名临时文件，多进程同时读写会导致文件冲突，返回错误结果。

最优实现方案 #

完全可以实现并行，核心逻辑是：每个迭代返回当前模拟对应的129维结果向量，通过foreach的combine参数自动拼接成最终结果矩阵，同时给每个迭代的临时文件加唯一后缀避免冲突。
你的需求天然符合并行条件：不同模拟层完全独立，同一层内的模型按类别串行运行即可满足相邻模型对比要求。

修改后的核心代码如下：

# 加载并行包，Windows用doParallel，Mac/Linux也可以用doMC
library(doParallel)
# 注册集群，核心数根据自己设备调整，建议留1-2个核心给系统
cl <- makeCluster(detectCores()-1)
registerDoParallel(cl)

inp <- array(1:(300*15*1500), dim=c(300,15,1500)) 
num_sims=1500

# 把需要的变量导出到worker节点（Windows必加，Mac/Linux可选）
clusterExport(cl, c("inp"), envir = environment())

# foreach通过combine=rbind把所有迭代返回的向量拼成矩阵
results <- foreach(i=1:num_sims, .packages=c('mclust','MplusAutomation'), .combine = rbind, .multicombine = TRUE) %dopar% {
  working <- inp[,,i]
  # 初始化当前迭代的结果向量，长度和预定义的129一致
  cur_res <- rep(NA, 129)
  
  cur_res[1] = working[1,17] #组别数
  cur_res[2] = working[1,18] #样本量1
  cur_res[3] = working[1,19] #样本量2
  cur_res[4] = working[1,20] #样本量3
  cur_res[5] = working[1,21] #dist2
  cur_res[6] = working[1,22] #dist3
  df <- as.data.frame(working[,1:15])
  # 生成当前迭代独有的临时文件名，避免冲突
  dat_file <- paste0("df_", i, ".dat")
  lpa1_file <- paste0("lpa1_15_", i, ".inp")
  lpa2_file <- paste0("lpa2_15_", i, ".inp")
  
  # 拟合1类模型
  lpa1_15 <- mplusObject(
    TITLE = "1-Class LPA;",
    VARIABLE = "USEVARIABLES = x01-x15;
     CLASSES=c(1);",
    ANALYSIS = "ESTIMATOR = MLR;
     TYPE=MIXTURE;",
    MODEL = "
     %OVERALL%
     x01-x15;
     [x01-x15];
     %c#1%
     x01-x15;
     [x01-x15];",
    usevariables = paste0("x", sprintf("%02d",1:15)),
    rdata = df)
  
  lpa1_15_fit = mplusModeler(lpa1_15, dat_file, modelout = lpa1_file, killOnFail = FALSE, run = 1L)
  
  if (!is.null(lpa1_15_fit$results$summaries$LL)){
    cur_res[7]  = -2 * lpa1_15_fit$results$summaries$LL
    cur_res[8]  = lpa1_15_fit$results$summaries$BIC
    cur_res[9]  = lpa1_15_fit$results$summaries$aBIC
    cur_res[10] = lpa1_15_fit$results$summaries$AIC
    cur_res[11] = lpa1_15_fit$results$summaries$AICC
  }
  # 拟合2类模型
  lpa2_15 <- mplusObject(
    TITLE = "2-Class LPA;",
    VARIABLE = "USEVARIABLES = x01-x15;
     CLASSES=c(2);",
    ANALYSIS = "ESTIMATOR = MLR;
     TYPE=MIXTURE;",
    MODEL = "
     %OVERALL%
     x01-x15;
     [x01-x15];
     %c#1%
     x01-x15;
    [x01-x15];
     %c#2%
     x01-x15;
     [x01-x15];",
    OUTPUT = "TECH11;",
    usevariables = paste0("x", sprintf("%02d",1:15)),
    rdata = df)
  
  lpa2_15_fit = mplusModeler(lpa2_15, dat_file, modelout = lpa2_file, killOnFail = FALSE, run = 1L)
  
  if (!is.null(lpa2_15_fit$results$summaries$LL)){
    cur_res[12] = -2 * lpa2_15_fit$results$summaries$LL
    cur_res[13] = lpa2_15_fit$results$summaries$BIC
    cur_res[14] = lpa2_15_fit$results$summaries$aBIC
    cur_res[15] = lpa2_15_fit$results$summaries$AIC
    cur_res[16] = lpa2_15_fit$results$summaries$AICC
    cur_res[17] = lpa2_15_fit$results$summaries$Entropy
    if (!is.null(lpa2_15_fit$results$summaries$T11_VLMR_2xLLDiff)){
      cur_res[18] = lpa2_15_fit$results$summaries$T11_VLMR_2xLLDiff
      cur_res[19] = lpa2_15_fit$results$summaries$T11_VLMR_PValue
      cur_res[20] = lpa2_15_fit$results$summaries$T11_LMR_Value
      cur_res[21] = lpa2_15_fit$results$summaries$T11_LMR_PValue
    }
  }
  
  # 其余类别模型拟合逻辑同理，按顺序写到cur_res对应位置即可
  
  # 关键：返回当前迭代的结果向量
  return(cur_res)
} 

# 关闭集群
stopCluster(cl)

补充注意事项 #

• 测试阶段可以先把num_sims设为2~3，确认结果格式、指标提取都正确后再跑全量1500次模拟。
• 跑完后可以手动删除生成的所有.dat、.inp、.out临时文件，也可以在迭代块最后加file.remove(dat_file, lpa1_file, lpa2_file...)自动清理。
• 如果需要保存中间结果避免跑崩丢失，可以每跑一批次就把当前results写入本地csv文件。

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