Пакет parallel и detectCores()
Запускайте кластеры на разветвлённых процессах или сокетах и распределяйте работу между ядрами CPU
«Пакет parallel и detectCores()» — бесплатный урок R Academy на CoddyKit. Это урок 1 из 4. Ты можешь прочитать весь урок бесплатно ниже — а потом практиковать его прямо в браузере с встроенным редактором кода и ИИ-репетитором 24/7. Это часть пути обучения R Academy, и твой прогресс синхронизируется между веб-версией и приложением CoddyKit. Курс R Academy содержит 4 уроков всего.
Зачем нужны параллельные вычисления
Современные компьютеры имеют несколько ядер CPU. По умолчанию R работает на одном ядре, оставляя остальные без нагрузки. Пакет parallel (встроенный в R) позволяет задействовать все ядра и ускорить повторяющиеся вычисления.
# 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
makeCluster(n) запускает n рабочих процессов. После завершения всегда вызывайте stopCluster(cl), чтобы освободить ресурсы. Распространённая практика — использовать detectCores() - 1, оставляя одно ядро для OS.
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: обмен переменными
Рабочие процессы имеют собственное пространство памяти и не видят ваше глобальное окружение. Используйте clusterExport(cl, varlist), чтобы скопировать именованные объекты из главного процесса во все рабочие процессы.
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: выполнение кода настройки
clusterEvalQ(cl, expr) вычисляет выражение в каждом рабочем процессе — это удобно для загрузки пакетов или подключения вспомогательных файлов на всех узлах до начала основных вычислений.
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
parLapply(cl, X, FUN) — параллельный аналог lapply(). Он распределяет элементы X между рабочими процессами и собирает результаты в список. Функция работает на всех платформах (Windows, macOS, Linux).
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: упрощённые результаты
parSapply(cl, X, FUN) — параллельная версия sapply(). Она автоматически упрощает результат до вектора или матрицы, когда это возможно, что удобно, если функция возвращает одно значение.
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: параллелизм на основе разветвления
mclapply() использует разветвление — копирует текущий процесс R вместо запуска новых, поэтому быстрее стартует, а рабочие процессы неявно получают доступ к окружению родительского процесса. Однако функция работает только в Unix/macOS, но не в Windows.
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 в параллельном режиме
Генерация случайных чисел в параллельном режиме требует осторожности — каждому рабочему процессу нужен независимый воспроизводимый поток. Используйте clusterSetRNGStream(cl, seed) с генератором L'Ecuyer-CMRG, чтобы получать воспроизводимые случайные последовательности.
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)Обработка ошибок в кластерах
Если рабочий процесс выдаёт ошибку, parLapply() повторно выдаёт её в главном процессе. Оберните вызовы в tryCatch() внутри функции или оберните весь вызов parLapply() в tryCatch(), чтобы корректно обрабатывать сбои.
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)Практический шаблон параллельных вычислений
Ниже приведён полный идиоматичный рабочий процесс параллельных вычислений: определить количество ядер, создать кластер, экспортировать зависимости, выполнить вычисления, собрать результаты и всегда остановить кластер — даже если произойдёт ошибка — с помощью on.exit().
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')Быстрая проверка
Какую функцию следует использовать, чтобы гарантировать вызов stopCluster(cl), даже если внутри параллельного кода произойдёт ошибка?
Итоги: пакет parallel
Основные выводы:
detectCores()сообщает количество доступных ядер; используйтеdetectCores() - 1для определения размера кластераmakeCluster(n)/stopCluster(cl)управляют жизненным циклом рабочих процессовclusterExport()копирует объекты;clusterEvalQ()выполняет код настройки в рабочих процессахparLapply()/parSapply()— кроссплатформенные параллельные итераторыmclapply()работает только в Unix, но быстрее запускается благодаря разветвлению- Используйте
on.exit(stopCluster(cl))для гарантированной очистки ресурсов - Параллелизм окупается только для вычислительно тяжёлых задач
# 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')Часто задаваемые вопросы
Урок «Пакет parallel и detectCores()» бесплатный?
Да — полный текст урока «Пакет parallel и detectCores()» бесплатно доступен здесь в веб-версии. Чтобы практиковать его интерактивно (встроенный редактор кода и ИИ-репетитор 24/7) и разблокировать остальной курс R Academy, подпишись на CoddyKit PRO. Курс R Academy содержит 4 уроков всего.
Чему я научусь в уроке «Пакет parallel и detectCores()»?
Запускайте кластеры на разветвлённых процессах или сокетах и распределяйте работу между ядрами CPU Ты практикуешь R Academy с помощью реального кода, который запускаешь прямо в браузере, и ИИ-репетитор 24/7 отвечает на твои вопросы во время урока.
Нужен ли мне опыт, чтобы начать R Academy?
Предыдущий опыт не требуется. R Academy на CoddyKit структурирован для всех уровней — от новичков до продвинутых, поэтому ты можешь начать отсюда или с самого начала и учиться в своем темпе. Это урок 1 из 4.
Сколько времени занимает урок «Пакет parallel и detectCores()»?
Большинство уроков CoddyKit занимают около 5–10 минут. Каждый из них компактный и интерактивный, поэтому ты постоянно делаешь прогресс и продолжаешь с того же места в веб-версии и приложении.
Можно ли писать и запускать код в этом уроке R Academy?
Да. Каждый урок R Academy включает встроенный редактор кода, поэтому ты пишешь и запускаешь реальный код прямо в браузере и получаешь моментальную обратную связь от AI — локальная установка не требуется.
Все уроки этого курса
- Пакет parallel и detectCores()
- Фреймворк future
- furrr: параллельные операции purrr
- Отладка и балансировка нагрузки параллельного кода