R Academy · 课时

调试并行代码与负载均衡

Q: 「调试并行代码与负载均衡」这节课中我会学到什么？

处理并行工作进程中的错误，并有效平衡不均匀的工作负载 你通过在浏览器中直接运行的动手代码来练习 R Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

处理并行工作进程中的错误，并有效平衡不均匀的工作负载

调试并行代码与负载均衡 是 CoddyKit 上的免费 R Academy 课时。这是第 4 节课，共 4 节。你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。这是 R Academy 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 R Academy 课程共包含 4 节课。

为什么并行调试很困难

并行代码很难调试，因为工作进程在独立的进程中运行：print() 输出对主进程不可见，browser() 等交互式调试器无法在工作进程内运行，而且错误会被序列化后在主进程中再次抛出，原始调用栈也会丢失。

library(parallel)

# Problem: cat() inside worker is invisible in the master
cl <- makeCluster(2)
clusterExport(cl, c())

result <- parLapply(cl, 1:4, function(i) {
  # This cat() goes to the worker's stdout, NOT the master
  cat('Worker processing', i, '\n')  # you will NOT see this
  i^2
})

cat('Master received:', unlist(result), '\n')
stopCluster(cl)

parLapply 中的错误传播

当工作进程抛出错误时，parLapply() 会将其包装后在主进程中再次抛出。错误消息会保留，但远程调用栈不会保留。请始终先顺序测试您的函数，再将其并行化。

library(parallel)
cl <- makeCluster(2)

# Step 1: test sequentially first!
my_fn <- function(x) {
  if (x == 3) stop('bad value: 3')
  sqrt(x)
}

# Sequential test reveals the bug before parallelising
tryCatch(
  lapply(1:5, my_fn),
  error = function(e) cat('Sequential error caught:', e$message, '\n')
)

# Fix: guard the bad case
my_fn_safe <- function(x) {
  if (x <= 0) return(NA_real_)
  sqrt(x)
}
clusterExport(cl, 'my_fn_safe')
cat(unlist(parLapply(cl, 1:5, my_fn_safe)), '\n')

stopCluster(cl)

工作进程函数内部的 tryCatch

将工作进程函数的函数体放在 tryCatch() 中，可以捕获并记录每个元素的错误，而不会导致整个并行任务崩溃。失败时请返回一个标记值（例如 NA），以便后处理识别出有问题的输入。

library(parallel)
cl <- makeCluster(2)

safe_compute <- function(x) {
  tryCatch({
    if (x == 3) stop('simulated error on element 3')
    list(value = x^2, error = NULL)
  }, error = function(e) {
    list(value = NA_real_, error = conditionMessage(e))
  })
}
clusterExport(cl, 'safe_compute')

results <- parLapply(cl, 1:5, safe_compute)

for (i in seq_along(results)) {
  r <- results[[i]]
  if (is.null(r$error)) {
    cat('Element', i, ': value =', r$value, '\n')
  } else {
    cat('Element', i, ': ERROR -', r$error, '\n')
  }
}

stopCluster(cl)

使用 tryCatch 调用 future::value()

在 future 包中，value(f) 会再次抛出远程错误。将其放在 tryCatch() 中，可以处理单个 future 的失败，同时继续收集其他 future 的结果。

library(future)
plan(multisession, workers = 2)

# Create futures — some will fail
inputs <- list(4, -1, 9, 'x', 16)

futures <- lapply(inputs, function(val) {
  future({
    if (!is.numeric(val)) stop('non-numeric input')
    if (val < 0) stop('negative input')
    sqrt(val)
  })
})

# Collect with individual error handling
results <- lapply(futures, function(f) {
  tryCatch(
    value(f),
    error = function(e) paste('ERROR:', e$message)
  )
})

for (i in seq_along(results)) {
  cat('Input:', inputs[[i]], '-> Result:', as.character(results[[i]]), '\n')
}

plan(sequential)

foreach 与 %dopar% 和 .errorhandling

foreach 包的 %dopar% 运算符会将迭代分发给已注册的后端。.errorhandling 参数控制发生错误时的行为：'stop'（默认，停止）、'remove'（跳过）或 'pass'（包含条件对象）。

library(foreach)
library(doParallel)

cl <- makeCluster(2)
registerDoParallel(cl)

# .errorhandling = 'pass': failed elements return the condition
results <- foreach(
  x = 1:6,
  .errorhandling = 'pass'
) %dopar% {
  if (x == 4) stop('bad element')
  x^2
}

for (i in seq_along(results)) {
  if (inherits(results[[i]], 'error')) {
    cat('Element', i, ': ERROR -', results[[i]]$message, '\n')
  } else {
    cat('Element', i, ': value =', results[[i]], '\n')
  }
}

stopCluster(cl)

负载均衡：静态与动态

静态调度会预先将大小相等的数据块分配给各个工作进程。动态调度则一次给每个工作进程分配一个任务，因此速度快的工作进程可以继续领取更多任务。当任务耗时差异很大时，动态调度效果更好。

library(parallel)
cl <- makeCluster(2)

# Simulate unequal task durations (element i takes i*0.05 seconds)
unequal_task <- function(i) {
  Sys.sleep(i * 0.05)
  i
}
clusterExport(cl, 'unequal_task')

# Static: parLapply distributes in fixed chunks
static_time <- system.time(
  parLapply(cl, 1:6, unequal_task)
)[['elapsed']]

# Dynamic: clusterApplyLB assigns one task per available worker
dynamic_time <- system.time(
  clusterApplyLB(cl, 1:6, unequal_task)
)[['elapsed']]

cat('Static LB: ', round(static_time, 2), 's\n')
cat('Dynamic LB:', round(dynamic_time, 2), 's\n')

stopCluster(cl)

分块：减少开销

进程间通信会为每个任务产生固定开销。处理许多小项目时，请将它们组合成更大的数据块，使每次工作进程调用通过一条消息完成更多工作，从而降低通信开销与计算量的比例。

library(parallel)
cl <- makeCluster(2)

# Naive: 1000 tiny tasks — high overhead
tiny_task <- function(x) x^2
clusterExport(cl, 'tiny_task')

t1 <- system.time(parLapply(cl, 1:1000, tiny_task))[['elapsed']]

# Chunked: 10 tasks of 100 items each — low overhead
chunk_task <- function(chunk) sapply(chunk, function(x) x^2)
chunks <- split(1:1000, cut(1:1000, 10, labels = FALSE))
clusterExport(cl, 'chunk_task')

t2 <- system.time(parLapply(cl, chunks, chunk_task))[['elapsed']]

cat('Unchunked:', round(t1, 4), 's\n')
cat('Chunked:  ', round(t2, 4), 's\n')

stopCluster(cl)

避免发送大型对象

将大型对象（数据框、矩阵、模型）序列化后发送给工作进程的成本很高。相反，请只传递索引，并从共享来源（文件、数据库或工作进程内预先加载的数据）读取数据。请尽量减小发送给工作进程的数据。

library(parallel)
cl <- makeCluster(2)

# BAD: sends the entire large data frame to every worker
big_df <- data.frame(x = rnorm(1e5), y = rnorm(1e5))
clusterExport(cl, 'big_df')  # 800 KB shipped to each worker

# BETTER: workers generate/read their own data slice
clusterEvalQ(cl, {
  set.seed(Sys.getpid())  # unique per worker
  local_data <- data.frame(x = rnorm(500), y = rnorm(500))
})

# Each worker uses its own local_data without receiving it from master
result <- parLapply(cl, 1:2, function(i) {
  coef(lm(y ~ x, data = local_data))
})
print(result)

stopCluster(cl)

从工作进程记录日志

由于工作进程无法写入主进程的控制台，请使用 makeCluster(outfile = '/path/to/log') 将工作进程的输出重定向到每个工作进程专用的日志文件。在 Unix 上，您也可以将输出重定向到 /dev/null，或为每个工作进程指定一个专用日志文件。

library(parallel)

# Route all worker stdout/stderr to a log file
log_file <- tempfile(fileext = '.log')
cl <- makeCluster(2, outfile = log_file)

clusterEvalQ(cl, {
  cat('[Worker', Sys.getpid(), '] started\n')
})

parLapply(cl, 1:4, function(i) {
  cat('[Worker] processing item', i, '\n')  # goes to log file
  i * 10
})

stopCluster(cl)

# Read the log
log_content <- readLines(log_file)
cat(head(log_content, 8), sep = '\n')

分析并行代码性能

使用 system.time() 测量总实际耗时，并手动分解工作进程开销与计算耗时。如需更详细的信息，请先使用 profvis::profvis() 顺序运行目标函数，然后再将瓶颈部分并行化。

library(parallel)

# Profile the task sequentially first
target_fn <- function(n) {
  x <- rnorm(n)
  list(
    mean = mean(x),
    sd   = sd(x),
    q95  = quantile(x, 0.95)
  )
}

# Measure sequential baseline
t_seq <- system.time(lapply(rep(1000, 20), target_fn))[['elapsed']]

# Measure parallel speedup
cl <- makeCluster(2)
clusterExport(cl, 'target_fn')
t_par <- system.time(parLapply(cl, rep(1000, 20), target_fn))[['elapsed']]
stopCluster(cl)

cat('Sequential:', round(t_seq, 4), 's\n')
cat('Parallel:  ', round(t_par, 4), 's\n')
cat('Efficiency:', round(t_seq / (t_par * 2) * 100, 1), '%\n')

完整的稳健模式

综合所有最佳实践：将工作分块，使用负载均衡分配，逐个元素处理错误，将日志记录到文件，并通过 on.exit() 确保集群得到清理。

library(parallel)

robust_parallel <- function(items, fn, workers = 2) {
  cl <- makeCluster(workers, outfile = tempfile())
  on.exit(stopCluster(cl), add = TRUE)

  clusterExport(cl, 'fn', envir = environment())

  safe_fn <- function(x) {
    tryCatch(
      fn(x),
      error = function(e) list(error = e$message, value = NA)
    )
  }
  clusterExport(cl, 'safe_fn', envir = environment())

  # Load-balanced assignment for unequal task times
  clusterApplyLB(cl, items, safe_fn)
}

results <- robust_parallel(1:8, function(x) {
  if (x == 5) stop('deliberate error')
  x^3
})

cat('Results:\n')
for (i in seq_along(results)) {
  cat(' [', i, ']', ifelse(is.na(results[[i]]$value),
    paste('ERROR:', results[[i]]$error),
    results[[i]]
  ), '\n')
}

快速检查

您有 100 个并行任务，它们的执行耗时差异很大（有些需要 10 毫秒，有些需要 500 毫秒）。哪种调度策略可以将总实际耗时降至最低？

回顾：调试并行代码

要点：

在并行化之前先顺序测试函数——先使用 lapply()
将工作进程函数体放在 tryCatch() 中，在不导致任务崩溃的情况下捕获每个元素的错误
使用带有 tryCatch() 的 future::value()，可以稳妥地处理单个 future 的失败
使用 foreach %dopar% 并将 .errorhandling = 'pass'，可以在结果列表中返回错误对象
使用 clusterApplyLB()，根据不等的任务耗时进行动态负载均衡
将小任务分块，以减少通信开销
尽量减少发送给工作进程的数据——传递索引，而不是大型数据对象
通过 makeCluster(outfile=) 将工作进程的输出记录到文件

# Canonical debugging workflow
# 1. Test sequentially
# lapply(inputs, my_fn)

# 2. Wrap in tryCatch
# safe_fn <- function(x) tryCatch(my_fn(x), error = function(e) NA)

# 3. Parallelise the safe version
# cl <- makeCluster(2); on.exit(stopCluster(cl))
# parLapply(cl, inputs, safe_fn)

cat('Debugging workflow: sequential -> tryCatch -> parallel\n')

常见问题解答

「调试并行代码与负载均衡」课时是免费的吗？

是的 — 「调试并行代码与负载均衡」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 R Academy 课程的其余内容，请升级到 CoddyKit PRO。 R Academy 课程共包含 4 节课。

「调试并行代码与负载均衡」这节课中我会学到什么？

处理并行工作进程中的错误，并有效平衡不均匀的工作负载你通过在浏览器中直接运行的动手代码来练习 R Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 R Academy 需要有经验吗？

无需任何先前经验。CoddyKit 上的 R Academy 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。这是第 4 节课，共 4 节。

「调试并行代码与负载均衡」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 R Academy 课中编写并运行代码吗？

能。每节 R Academy 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

← 返回 R Academy