แพ็กเกจ parallel และ detectCores()

เหตุใดจึงต้องใช้การประมวลผลแบบขนาน

# Check how many cores your machine has
library(parallel)

total_cores <- detectCores()
logical_cores <- detectCores(logical = TRUE)
physical_cores <- detectCores(logical = FALSE)

cat('Total logical cores:', total_cores, '
')
cat('Physical cores:', physical_cores, '
')

makeCluster และ stopCluster

library(parallel)

# Spawn workers (leave 1 core for system)
n_cores <- detectCores() - 1
cl <- makeCluster(n_cores)

cat('Cluster created with', n_cores, 'workers\n')

# Always clean up!
stopCluster(cl)
cat('Cluster stopped.\n')

clusterExport: การแบ่งปันตัวแปร

library(parallel)
cl <- makeCluster(2)

# Define a variable and a function in the master
base_value <- 100
add_base <- function(x) x + base_value

# Export them to workers
clusterExport(cl, varlist = c('base_value', 'add_base'))

# Now workers can use them
result <- parLapply(cl, 1:4, function(x) add_base(x))
cat(unlist(result), '\n')  # 101 102 103 104

stopCluster(cl)

clusterEvalQ: การเรียกใช้โค้ดตั้งค่า

library(parallel)
cl <- makeCluster(2)

# Load a package on every worker
clusterEvalQ(cl, {
  library(stats)
  set.seed(42)  # set seed per worker
})

# Each worker can now use stats functions
result <- parLapply(cl, 1:4, function(n) rnorm(n, mean = 0, sd = 1))
result[[1]]  # one random normal value

stopCluster(cl)

parLapply: lapply แบบขนาน

library(parallel)
cl <- makeCluster(2)

# Simulate slow computation: sleep 0.1s per item
clusterEvalQ(cl, Sys.sleep)

slow_square <- function(x) {
  Sys.sleep(0.05)
  x^2
}

clusterExport(cl, 'slow_square')

system.time(
  result <- parLapply(cl, 1:8, slow_square)
)
cat(unlist(result), '\n')  # 1 4 9 16 25 36 49 64

stopCluster(cl)

parSapply: ผลลัพธ์แบบย่อ

library(parallel)
cl <- makeCluster(2)

# parSapply returns a simplified vector
squares <- parSapply(cl, 1:10, function(x) x^2)
cat(squares, '\n')  # 1 4 9 16 25 36 49 64 81 100

# Returns a matrix if FUN returns a vector of same length
stats_result <- parSapply(cl, 1:4, function(n) {
  x <- rnorm(100)
  c(mean = mean(x), sd = sd(x))
})
print(stats_result)  # 2x4 matrix

stopCluster(cl)

mclapply: การประมวลผลแบบขนานด้วยการแยกกระบวนการ

library(parallel)

# mclapply: Unix/macOS only
# Workers inherit the parent environment automatically
base_value <- 42

if (.Platform$OS.type != 'windows') {
  result <- mclapply(
    1:8,
    function(x) x * base_value,  # base_value visible without export
    mc.cores = 4
  )
  cat(unlist(result), '\n')
} else {
  cat('mclapply not supported on Windows. Use parLapply instead.\n')
}

การวัดประสิทธิภาพแบบลำดับเทียบกับแบบขนาน

library(parallel)
cl <- makeCluster(2)

# Task: compute 100 iterations of matrix multiply
heavy_task <- function(n) {
  m <- matrix(rnorm(200), nrow = 100)
  sum(m %*% t(m))
}
clusterExport(cl, 'heavy_task')

seq_time <- system.time(lapply(1:20, heavy_task))[['elapsed']]
par_time <- system.time(parLapply(cl, 1:20, heavy_task))[['elapsed']]

cat('Sequential:', round(seq_time, 3), 's\n')
cat('Parallel:  ', round(par_time, 3), 's\n')
cat('Speedup:   ', round(seq_time / par_time, 2), 'x\n')

stopCluster(cl)

การกำหนดค่าเริ่มต้น RNG ในการประมวลผลแบบขนาน

library(parallel)
cl <- makeCluster(2)

# Set reproducible RNG streams across workers
clusterSetRNGStream(cl, iseed = 123)

# Each worker uses its own independent random stream
results1 <- parSapply(cl, 1:6, function(i) rnorm(1))

# Reset and repeat — same results
clusterSetRNGStream(cl, iseed = 123)
results2 <- parSapply(cl, 1:6, function(i) rnorm(1))

cat('Run 1:', round(results1, 4), '\n')
cat('Run 2:', round(results2, 4), '\n')
cat('Identical:', identical(results1, results2), '\n')

stopCluster(cl)

การจัดการข้อผิดพลาดในคลัสเตอร์

library(parallel)
cl <- makeCluster(2)

# Wrap risky code inside the worker function
safe_log <- function(x) {
  tryCatch(
    log(x),
    warning = function(w) NA_real_,
    error   = function(e) NA_real_
  )
}
clusterExport(cl, 'safe_log')

# -1 produces NaN warning, 'a' produces an error
input <- list(4, 9, -1, 'a', 16)
result <- parLapply(cl, input, safe_log)
cat(unlist(result), '\n')  # 1.386 2.197 NaN NA 2.773

stopCluster(cl)

รูปแบบการประมวลผลแบบขนานที่ใช้งานได้จริง

library(parallel)

run_parallel <- function(data, fn, n_workers = detectCores() - 1) {
  cl <- makeCluster(n_workers)
  on.exit(stopCluster(cl))  # guaranteed cleanup

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

  result <- parLapply(cl, data, fn)
  result
}

# Example: bootstrap mean estimation
samples <- lapply(1:100, function(i) rnorm(50, mean = 5, sd = 2))
means <- run_parallel(samples, mean)

cat('Grand mean:', round(mean(unlist(means)), 3), '\n')
cat('95% CI: [',
    round(quantile(unlist(means), 0.025), 3), ',',
    round(quantile(unlist(means), 0.975), 3), ']\n')

ตรวจสอบด่วน

ทบทวน: แพ็กเกจ parallel

# Minimal reproducible parallel pattern
library(parallel)

cl <- makeCluster(max(1, detectCores() - 1))
on.exit(stopCluster(cl))

clusterSetRNGStream(cl, iseed = 42)
result <- parSapply(cl, 1:8, function(x) x^2 + rnorm(1, 0, 0.1))
cat(round(result, 2), '\n')