使用parLapply并行化时已导出对象market_time无法识别的问题

使用parLapply并行处理时，已尝试将必要的data.table对象dt及相关变量导出到计算核心，但运行时报错找不到market_time（该对象是dt的一列）。若改为单独导出dt的每一列，则会报错找不到函数内计算的jacobian对象。

主代码 #

library('data.table')
library('numDeriv')
library('snow')

cores=detectCores()
cl <- makeCluster(cores[1], type = 'PSOCK')

markets <- unique(dt[, market_time])
R = 10000
nu_p <- rnorm(n = R, -2, 0.5)
nu_xr <- rnorm(n = R, 2, 0.5)
nu_xm <- rnorm(n = R, 2, 0.5)
nu_xj <- rnorm(n = R, 2, 0.5)

clusterExport(cl,c('dt','nu_p','nu_xr','nu_xm','nu_xj')) # 原代码缺少右括号，已修正

temp <- parLapply(cl, markets,calc_mc_w, dt=dt,nu_p=nu_p,nu_xr= nu_xr,
                 nu_xm=nu_xm,nu_xj=nu_xj)

自定义函数定义 #

calc_mc_w <- function(m, dt,nu_p,nu_xr,nu_xm,nu_xj){
  dt_mkt = dt[market_time==m,]
  market_time <- dt_mkt[, market_time]
  x_m <- dt_mkt[, x_m]
  x_j <- dt_mkt[, x_j]
  x_r <- dt_mkt[, x_r]
  p <- as.matrix(dt_mkt[, p])
  xi <- dt_mkt[, xi]
  p <- as.matrix(dt_mkt[, p])
  jacobian <- jacobian(function(x){calc_shares(x, x_m, x_j, x_r, xi, nu_p, 
                                               nu_xm, nu_xj, nu_xr, 
                                               market_time)},p)
  output <- dt_mkt[,c('prod','market','time','retailer')]
  
  # 构建与未知量数量匹配的方程组
  retailers = unique(dt_mkt[, retailer])
  
  temp <- lapply(retailers,calc_mc_w_r,dt_mkt = dt_mkt, jacobian = jacobian)
  
  temp <- rbindlist(temp)
  output <- merge(output,temp,by.x = c('prod','retailer'),
                  by.y = c('prod','retailer'), allow.cartesian=TRUE)
  output
}

calc_mc_w_r <- function(r, dt_mkt, jacobian){
  dt_r = dt_mkt[retailer == r,]
  result <- dt_r[,c('prod','retailer')]
  rows = (dt_mkt[,'retailer']== r)
  jacobian_r = jacobian[rows,rows]
  result <- result[,mc_w := solve(jacobian_r, dt_r[,shares]+ jacobian_r %*% dt_r[,p])]
  result
}

错误信息 #

Error in checkForRemoteErrors(val) :
2 nodes produced errors; first error: object 'market_time' not found

解决方法 #

1. 修复clusterExport的语法错误

原代码中clusterExport语句缺少右括号，导致对象根本没有成功导出到集群节点。修正后才能确保dt等变量被正确传递：

clusterExport(cl,c('dt','nu_p','nu_xr','nu_xm','nu_xj'))

2. 导出所有依赖函数到集群节点

calc_mc_w内部调用了calc_shares和calc_mc_w_r，这些函数没有被导出到集群节点，导致并行计算时无法找到。需要将这些函数也导出：

clusterExport(cl,c('dt','nu_p','nu_xr','nu_xm','nu_xj', 'calc_mc_w_r', 'calc_shares'))

或者用clusterEvalQ加载函数定义：

clusterEvalQ(cl, {
  library(data.table)
  library(numDeriv)
  # 在此复制calc_mc_w_r和calc_shares的函数定义
})

3. 解决匿名函数的变量作用域问题

jacobian中的匿名函数引用了外部环境的变量（如market_time、nu_p等），在并行环境下可能因作用域丢失找不到变量。可以通过显式传递变量强制绑定：

# 修改jacobian调用部分，将变量作为参数传入匿名函数
jacobian <- jacobian(function(x, x_m, x_j, x_r, xi, nu_p, nu_xm, nu_xj, nu_xr, market_time){
  calc_shares(x, x_m, x_j, x_r, xi, nu_p, nu_xm, nu_xj, nu_xr, market_time)
}, p, x_m=x_m, x_j=x_j, x_r=x_r, xi=xi, nu_p=nu_p, nu_xm=nu_xm, nu_xj=nu_xj, nu_xr=nu_xr, market_time=market_time)

4. 优化：提前拆分数据集减少传输开销

不需要将整个dt传递给每个节点，提前按market_time拆分数据集，每个节点只处理对应子集，减少内存占用和数据传输：

# 提前按market_time分组拆分dt
dt_list <- split(dt, by = 'market_time')
# 直接传递子集列表给parLapply，无需再传整个dt
temp <- parLapply(cl, dt_list, calc_mc_w_modified, nu_p=nu_p, nu_xr=nu_xr, nu_xm=nu_xm, nu_xj=nu_xj)
# 修改calc_mc_w函数，直接接收子集dt_mkt
calc_mc_w_modified <- function(dt_mkt, nu_p, nu_xr, nu_xm, nu_xj){
  market_time <- dt_mkt[, market_time]
  # 后续逻辑保持不变...
}

内容的提问来源于stack exchange，提问作者Jakob D.