Привет, Хабр!
Параллельные вычисления – подход к проектированию и выполнению программ, который позволяет ускорить обработку данных и вычисления, используя множество процессоров или ядер процессора одновременно.
В ЯП R паралельное выполнение также имеет свои варианты реализации. Рассмотрим их в статье.
Немного о том, что такое параллельное программирование в общих чертах
Существует два основных подхода к параллельному программированию: использование потоков и использование процессов.
На основе потоков задачи разделяются на независимые подзадачи, которые могут выполняться одновременно. Каждая подзадача выполняется в соответствующем потоке, который может работать параллельно с другими потоками. В этом случае все потоки имеют доступ к общей памяти и могут обмениваться данными между собой.
А на основе процессов, задачи разделяются на отдельные процессы, которые выполняются независимо друг от друга. Каждый процесс имеет свою собственную память, и обмен данными между процессами может выполняться с использованием механизмов межпроцессного взаимодействия, таких как сокеты или каналы.
Основные варианты реализации в R и пакеты
parallel
parallel объединяет функциональность более старых пакетов multicore и snow, предоставляя интерфейс для выполнения задач в параллельном режиме. Основные функции:
mclapply(): аналог функции lapply(), используемый для параллельной обработки списка элементов. Работает только в юниксо подобных ос:
library(parallel) result <- mclapply(1:100, function(x) x^2, mc.cores = 2)parLapply(): альтернатива lapply(), подходящая для работы как на одном пк, так и на кластере. Для использования необходимо предварительно создать кластер с помощью makeCluster():
cl <- makeCluster(4) # cоздаем кластер из 4 узлов clusterExport(cl, varlist = "myData") # экспортируем данные в каждый узел кластера result <- parLapply(cl, 1:100, function(x) x^2) stopCluster(cl)makeCluster(): создает кластер из заданного количества ядер. Кластер затем используется функциями типа parLapply():
cl <- makeCluster(4) # 4 ядра stopCluster(cl)clusterApply(): распараллеливает выполнение функции по элементам списка на кластере:
cl <- makeCluster(2) result <- clusterApply(cl, 1:10, function(x) x^2) stopCluster(cl)clusterEvalQ() и clusterCall(): позволяют выполнить произвольный код на узлах кластера:
cl <- makeCluster(2) clusterEvalQ(cl, library(ggplot2)) # ggplot2 на каждом узле clusterCall(cl, function() Sys.info()['nodename']) # получаем имя узла stopCluster(cl)Попробуем распараллелить анализ данных с использованием модели линейной регрессии на различных подвыборках датасета:
library(parallel) data(mtcars) cl <- makeCluster(4) # разделяем данные на подвыборки subsets <- split(mtcars, mtcars$cyl) # распараллеливаем вычисления models <- parLapply(cl, subsets, function(subset) lm(mpg ~ wt, data = subset)) stopCluster(cl)Код создает параллельный кластер, разделяет набор данных mtcars на подвыборки по количеству цилиндров, и использует функцию parLapply() для параллельного подсчета линейных моделей по каждой подвыборке.
foreach
Пакет foreach предоставляет способ организации циклов в R, который может быть легко адаптирован для параллельной обработки без необходимости изменения основного кода. Основная фича заключается в том, чтобы заменить традиционные циклы R, такие как for, на конструкцию foreach, которая затем может использоваться с одним из многих пакетов параллельного бэкенда, к примеруdoParallel.
Основные функции foreach:
foreach(): основная функция, которая используется для определения цикла. Без паралельной обработки:
library(foreach) result <- foreach(i = 1:100) %do% { sqrt(i) }doParallel является одним из бэкендов для foreach, он позволяет выполнять каждую итерацию цикла foreach параллельно, используя множество процессоров или ядер на компьютере. Основные функции:
registerDoParallel(): функция для регистрации doParallel как параллельного бэкенда для foreach, указывая количество ядер или процессоров, которые должны использоваться для параллельной обработки.
Пример использования обоих пакетов для параллельной обработки:
library(foreach) library(doParallel) cl <- makeCluster(4) # создаем кластер из 4 ядер registerDoParallel(cl) # регистрируем кластер для использования с foreach result <- foreach(i = 1:100) %dopar% { sqrt(i) } stopCluster(cl)Здесь создали кластер из четырех ядер с помощью makeCluster из пакета parallel на который doParallel опирается и затем регистрируем этот кластер как бэкенд для foreach, используя registerDoParallel. Цикл foreach автоматически распараллеливается, и каждая итерация выполняется на отдельном ядре.
foreach также поддерживают различные механизмы контроля за процессом выполнения, например агрегацию результатов с помощью аргумента .combine, который позволяет определить, как результаты отдельных итераций должны быть объединены в конечный resultd:
result <- foreach(i=1:100, .combine='c') %dopar% { sqrt(i) }Результаты каждой итерации объединяются в один вектор с использованием функции объединения 'c'.
future
plan(): функция определяет, как будут выполняться наши future-объекты — локально, в параллельных сессиях, на удаленных машинах и т.д. выбр Можно выбраать стратегию исполнения, например, multisession, multicore, sequential и другие.
future(): создает future-объект, представляющий значение, которое будет доступно в будущем. Это сердце пакета, позволяющее нам абстрагироваться от деталей параллельного исполнения.
value(): получает результат выполнения future-объекта, дожидаясь его завершения, если это необходимо.
resolved(): проверяет, завершилось ли выполнение future-объекта.
future и plan для параллельного вычисления:
library(future) plan(multisession) # несколько сессий для параллельного исполнения # future-объект для асинхронного вычисления fut <- future({ Sys.sleep(5) # имитация длительной операции sum(1:1000) }) result <- value(fut)Инициируем параллельное вычисление суммы чисел от 1 до 1000, при этом основной поток исполнения не блокируется и может продолжать выполнение других задач.
Можно так же реализовать веб.скрапинг к примеру для сбора информации о продуктах с нескольких интернет-магазинов для сравнения цен. можно одновременно отправлять запросы к разным сайтам, уменьшая общее время выполнения скрапинга:
library(future) plan(multisession) urls <- c("https://store1.com/product", "https://store2.com/product", "https://store3.com/product") scrapingFutures <- lapply(urls, function(url) { future({ scrapeData(url) # псевдофункция для извлечения данных со страницы }) }) productsData <- lapply(scrapingFutures, value)RcppParallel
RcppParallel использует потоки C++ и предлагает простой API для наиболее распространенных операций параллельного программирования в c++.
Допустим есть большой вектор чисел, и мы хотим быстро посчитать его сумму. Используем RcppParallel для буста процесса:
#include <RcppParallel.h> using namespace RcppParallel; struct Sum : public Worker { // вектор и результат const RVector<double> input; double& sum; // конструктор Sum(const NumericVector input, double& sum) : input(input), sum(sum) {} // переопределение оператора () void operator()(std::size_t begin, std::size_t end) { sum += std::accumulate(input.begin() + begin, input.begin() + end, 0.0); } }; // [[Rcpp::export]] double parallelSum(NumericVector x) { double sum = 0; Sum sumWorker(x, sum); parallelReduce(0, x.length(), sumWorker); return sum; }Структура Sum используется для параллельного вычисления суммы чисел в векторе. parallelReduce автоматически разбивает данные на части и выполняет их параллельно, используя доступные ядра процессора.
Предположим, нам нужно применить функцию к каждому элементу большого вектора. Вместо того, чтобы делать это последовательно, мы можем выполнить операцию параллельно:
#include <RcppParallel.h> using namespace RcppParallel; struct Transform : public Worker { // исходный и целевой векторы const RVector<double> input; RVector<double> output; Transform(const NumericVector input, NumericVector output) : input(input), output(output) {} // переопределение оператора () void operator()(std::size_t begin, std::size_t end) { std::transform(input.begin() + begin, input.begin() + end, output.begin() + begin, [](double x) { return std::sqrt(x); // Например, вычисляем корень квадратный }); } }; // [[Rcpp::export]] NumericVector parallelTransform(NumericVector x) { NumericVector result(x.size()); Transform transformWorker(x, result); parallelFor(0, x.size(), transformWorker); return result; }Струтура Transform применяет некоторую функцию к элементам вектора параллельно.
Прочие примеры применения
Операции с матрицами порой оч. больших вычислительных ресурсов, особенно при работе с большими матрицами:
library(foreach) library(doParallel) # задаем количество ядер cl <- makeCluster(detectCores()) registerDoParallel(cl) # генерим две большие матрицы mat1 <- matrix(rnorm(1e6), 1000, 1000) mat2 <- matrix(rnorm(1e6), 1000, 1000) # параллельное умножение матриц result <- foreach(i=1:nrow(mat1), .combine='rbind') %dopar% { mat1[i, ] %*% mat2 } stopCluster(cl)Как можно параллельно обработать большой датасет:
library(foreach) library(doParallel) cl <- makeCluster(detectCores()) registerDoParallel(cl) # представим, что у есть большой датасет 'big_data' # используем placeholder для демонстрации big_data <- replicate(100, rnorm(1e4)) # применяем функцию обработки данных к каждому столбцу датасета processed_data <- foreach(i=1:ncol(big_data), .combine='cbind') %dopar% { someComplexFunction(big_data[, i]) } stopCluster(cl)Как можно параллельно выполнять множество независимых вычислительных задач:
library(foreach) library(doParallel) cl <- makeCluster(detectCores()) registerDoParallel(cl) # выполнение независимых вычислений, например, генерация симуляций tasks <- 100 # Количество задач results <- foreach(i=1:tasks, .combine='c') %dopar% { runSimulation(i) # 'runSimulation' - placeholder для функции симуляции } stopCluster(cl)В целом параллельное вычисление в R реализовано добротно, однако еще далеко до питона.
Статья подготовлена в рамках запуска специализации Системный аналитик и бизнес-аналитик. По ссылкам ниже вы сможете зарегистрироваться на бесплатные вебинары специализации.
ссылка на оригинал статьи https://habr.com/ru/articles/791194/
Добавить комментарий