Data Processing and Analysis with {doParallel}

The {doParallel} package in R is a powerful tool for parallel computing, enabling users to execute tasks concurrently to improve computational efficiency, especially for time-consuming operations. It serves as a backend for the {foreach} package, providing a mechanism to distribute computations across multiple CPU cores or processors, thereby reducing execution time for tasks that can be parallelized.

This package is part of the R ecosystem for parallel processing. It integrates the {foreach} package, which provides a looping construct for executing iterative tasks, with the {parallel} package, a base R package introduced in R 2.14.0 for parallel computation. {doParallel} simplifies the process of running loops in parallel by leveraging multiple CPU cores, making it particularly useful for tasks like simulations, large-scale data processing, or computationally intensive statistical analyses.

Key features of {doParallel}:

  • Parallel Execution: Distributes loop iterations across multiple cores or processors.
  • Ease of Use: Works seamlessly with {foreach}, allowing users to write parallel loops with minimal code changes compared to standard sequential loops.
  • Cross-Platform Support: Supports both Windows and Unix-like systems (e.g., Linux, macOS) through different parallelization mechanisms.
  • Scalability: Can utilize multiple cores on a single machine, making it accessible for users without access to a computing cluster.

How {doParallel} Works

{doParallel} combines the {foreach} package’s looping framework with the {parallel} package’s ability to manage parallel workers. It sets up a parallel backend that allows {foreach} loops to execute iterations concurrently across multiple CPU cores. Below is a step-by-step explanation of how it works:

Check and Install Required R Packages

Code 
packages<- c(
'tidyverse',
'data.table',
'feather',
'arrow',
'doParallel',
'foreach'
)
#| warning: false
#| error: false

# Install missing packages
new_packages <- packages[!(packages %in% installed.packages()[,"Package"])]
if(length(new_packages)) install.packages(new_packages)

Verify Instalation

Code 
# Verify 
cat("Installed packages:\n")
Installed packages:
Code 
print(sapply(packages, requireNamespace, quietly = TRUE))
 tidyverse data.table    feather      arrow doParallel    foreach 
      TRUE       TRUE       TRUE       TRUE       TRUE       TRUE

Load R-packages

Code 
# Load packages with suppressed messages
invisible(lapply(packages, function(pkg) {
  suppressPackageStartupMessages(library(pkg, character.only = TRUE))
}))

Check Loaded Packages

Code 
# Check loaded packages
cat("Successfully loaded packages:\n")
Successfully loaded packages:
Code 
print(search()[grepl("package:", search())])
 [1] "package:doParallel" "package:parallel"   "package:iterators" 
 [4] "package:foreach"    "package:arrow"      "package:feather"   
 [7] "package:data.table" "package:lubridate"  "package:forcats"   
[10] "package:stringr"    "package:dplyr"      "package:purrr"     
[13] "package:readr"      "package:tidyr"      "package:tibble"    
[16] "package:ggplot2"    "package:tidyverse"  "package:stats"     
[19] "package:graphics"   "package:grDevices"  "package:utils"     
[22] "package:datasets"   "package:methods"    "package:base"

Set Up the Parallel Backend

{doParallel} uses the {parallel} package to create a cluster of worker processes. You specify the number of CPU cores to use, and {doParallel} registers these as a parallel backend for {foreach}.

Code 
# Detect the number of available CPU cores
cores <- detectCores() - 1  # Leave one core free to avoid overloading
# Register the parallel backend
registerDoParallel(cores)
  • On Unix-like systems, {doParallel} uses “forking” (via mclapply) to create parallel processes.
  • On Windows, it uses “PSOCK” (Parallel Socket Cluster) to spawn separate R processes for parallel execution.

Write a {foreach} Loop

The {foreach} package provides a looping construct similar to a for loop but designed for parallel execution. You write the loop using %dopar% to indicate that iterations should run in parallel.

Code 
# Example: Parallel computation of squares
result <- foreach(i = 1:10, .combine = c) %dopar% {
i^2  # Each iteration computes the square of i
}
  • .combine = c: Specifies how to combine results from parallel iterations (here, results are concatenated into a vector).
  • %dopar%: Instructs {foreach} to execute the loop in parallel using the registered {doParallel} backend.

Execute and Collect Results

Each iteration of the {foreach} loop is distributed to a worker process (or core). The results are collected and combined as specified (e.g., into a vector, list, or matrix). In the example above, result will contain the squares of numbers 1 to 10: [1, 4, 9, 16, 25, 36, 49, 64, 81, 100].

Clean Up

After completing the parallel tasks, it’s good practice to stop the cluster to free up system resources:

Code 
stopImplicitCluster()

Example: Parallel Matrix Multiplication

Here’s a practical example demonstrating {doParallel} for matrix multiplication across multiple cores:

Code 
# Set up parallel backend
cores <- detectCores() - 1
registerDoParallel(cores)

# Create sample matrices
n <- 1000
mat1 <- matrix(rnorm(n * n), nrow = n)
mat2 <- matrix(rnorm(n * n), nrow = n)

# Parallel matrix multiplication
result <- foreach(i = 1:nrow(mat1), .combine = rbind) %dopar% {
  row <- mat1[i, ]
  apply(mat2, 1, function(col) sum(row * col))
}

# Stop the cluster
stopImplicitCluster()

# View result
dim(result)  # Should be n x n
[1] 1000 1000

Key Components and Options

  • Parallel Backend: {doParallel} relies on the {parallel} package’s makeCluster() (for Windows) or mclapply() (for Unix-like systems) to manage parallel workers.

  • .combine Argument: Controls how results are aggregated (e.g., c for vectors, rbind for matrices, list for lists).

  • .packages Argument: Specifies additional packages to load in each worker process if needed (e.g., .packages = "dplyr").

  • Error Handling: Use .errorhandling = "remove" or "stop" to handle errors in parallel iterations gracefully.

How It Improves Performance

By distributing iterations across multiple cores, {doParallel} reduces execution time for tasks that are “embarrassingly parallel” (i.e., iterations are independent). For example, if a loop has 100 iterations and you have 4 cores, {doParallel} can theoretically process 25 iterations per core simultaneously, potentially reducing runtime by up to 75% (though overhead like communication between workers may reduce this gain).

Limitations

  • Overhead: Creating and managing worker processes introduces some overhead, so parallelization is only beneficial for computationally intensive tasks.
  • Memory Usage: Each worker process may duplicate data, increasing memory demands.
  • Windows Constraints: Windows uses PSOCK clusters, which are slower than forking on Unix-like systems due to process creation overhead.
  • Dependencies: Tasks must be independent, as {doParallel} does not handle inter-iteration dependencies well.

Parallel Data Processing and Analysis with {doParallel}

This section of tutorial demonstrates how to use the {doParallel} package in R for parallel computing applied to data processing and basic statistical analyses on a large dataset. We’ll use the NYC Yellow Taxi Trip Records for January 2023, stored in the Parquet file yellow_tripdata_2023-01.parquet. This dataset contains approximately 3 million rows of taxi trip data, making it suitable for showcasing parallel processing to speed up computations.

Data

Code 
# read with read_csv()
dataFolder<-"/home/zia207/WebSites/R_Website/High_Performance_Computing_R/R_Markdown/Data/"
# Load January 2023 data
df <- read_parquet(paste0(dataFolder, "yellow_tripdata_2023-01.parquet"))

Setting Up Parallel Backend

Detect available cores and register the parallel backend. We’ll use all but one core to avoid system overload.

Code 
cores <- detectCores() - 1
registerDoParallel(cores)

Data Processing with {doParallel}

For large datasets, parallel processing can accelerate data cleaning and feature engineering. Here, we’ll add a new column for trip duration (in minutes) and filter out invalid rows (e.g., negative fares or distances). To demonstrate parallelism, we’ll split the data into chunks, process each chunk in parallel, and combine the results.

First, split the data into a list of chunks (e.g., 4 chunks for simplicity).

Code 
num_chunks <- 4
chunk_size <- floor(nrow(df) / num_chunks)  # Use floor to ensure integer division
chunks <- lapply(1:num_chunks, function(i) {
  start <- as.integer((i - 1) * chunk_size + 1)
  end <- if (i == num_chunks) nrow(df) else as.integer(i * chunk_size)
  df[start:end, ]
})

Now, use {foreach} with %dopar% to process each chunk in parallel: calculate trip duration, filter invalid data (trip_distance > 0, fare_amount > 0, passenger_count > 0).

Code 
processed_chunks <- foreach(chunk = chunks, .combine = rbind, .packages = "dplyr") %dopar% {
  chunk %>%
    mutate(trip_duration = as.numeric(difftime(tpep_dropoff_datetime, tpep_pickup_datetime, units = "mins"))) %>%
    filter(trip_distance > 0, fare_amount > 0, passenger_count > 0, trip_duration > 0)
}

dim(processed_chunks)  # Check dimensions after processing
[1] 2884165      20
Code 
head(processed_chunks)
# A tibble: 6 × 20
  VendorID tpep_pickup_datetime tpep_dropoff_datetime passenger_count
     <int> <dttm>               <dttm>                          <dbl>
1        2 2022-12-31 19:32:10  2022-12-31 19:40:36                 1
2        2 2022-12-31 19:55:08  2022-12-31 20:01:27                 1
3        2 2022-12-31 19:25:04  2022-12-31 19:37:49                 1
4        2 2022-12-31 19:10:29  2022-12-31 19:21:19                 1
5        2 2022-12-31 19:50:34  2022-12-31 20:02:52                 1
6        2 2022-12-31 19:09:22  2022-12-31 19:19:49                 1
# ℹ 16 more variables: trip_distance <dbl>, RatecodeID <dbl>,
#   store_and_fwd_flag <chr>, PULocationID <int>, DOLocationID <int>,
#   payment_type <int>, fare_amount <dbl>, extra <dbl>, mta_tax <dbl>,
#   tip_amount <dbl>, tolls_amount <dbl>, improvement_surcharge <dbl>,
#   total_amount <dbl>, congestion_surcharge <dbl>, airport_fee <dbl>,
#   trip_duration <dbl>

This parallel approach distributes the mutate and filter operations across cores, speeding up processing for large datasets.

Descriptive Statistics

Descriptive statistics like mean, median, min, max, and standard deviation can be computed on subsets or via bootstrapping in parallel for efficiency and to estimate variability.

Example: Compute bootstrap means for key variables (trip_distance, fare_amount, tip_amount) in parallel. We’ll perform 100 bootstrap resamples per variable.

Code 
variables <- c("trip_distance", "fare_amount", "tip_amount")
boot_results <- foreach(var = variables, .combine = rbind) %dopar% {
  boot_means <- replicate(100, mean(sample(processed_chunks[[var]], replace = TRUE), na.rm = TRUE))
  data.frame(variable = var, mean = mean(boot_means), sd = sd(boot_means))
}

boot_results
       variable      mean          sd
1 trip_distance  3.474734 0.013134137
2   fare_amount 18.548502 0.009567273
3    tip_amount  3.400448 0.002313311

This outputs a data frame with the average bootstrap mean and standard deviation for each variable, computed in parallel across variables.

For a more granular approach, compute summary statistics on grouped data (e.g., by payment_type) in parallel.

Code 
payment_types <- unique(processed_chunks$payment_type)
group_stats <- foreach(pt = payment_types, .combine = rbind, .packages = "dplyr") %dopar% {
  subset <- filter(processed_chunks, payment_type == pt)
  data.frame(
    payment_type = pt,
    mean_distance = mean(subset$trip_distance, na.rm = TRUE),
    median_fare = median(subset$fare_amount, na.rm = TRUE),
    sd_duration = sd(subset$trip_duration, na.rm = TRUE)
  )
}

group_stats
  payment_type mean_distance median_fare sd_duration
1            2      3.472557        12.1    53.63800
2            1      3.476022        12.8    40.60209
3            4      3.260377        12.1    23.77467
4            3      3.036158        10.0    20.17344

Correlation Analysis

For correlations between multiple pairs of variables, parallelize the computation, especially useful for large matrices or bootstrapped correlations.

Example: Compute Pearson correlations between selected numeric variables in parallel.

Code 
num_vars <- c("trip_distance", "fare_amount", "tip_amount", "total_amount", "trip_duration")
cor_pairs <- combn(num_vars, 2)  # All unique pairs

cor_results <- foreach(i = 1:ncol(cor_pairs), .combine = rbind) %dopar% {
  var1 <- cor_pairs[1, i]
  var2 <- cor_pairs[2, i]
  corr <- cor(processed_chunks[[var1]], processed_chunks[[var2]], use = "complete.obs")
  data.frame(var1 = var1, var2 = var2, correlation = corr)
}

cor_results
            var1          var2 correlation
1  trip_distance   fare_amount  0.19447138
2  trip_distance    tip_amount  0.12022975
3  trip_distance  total_amount  0.19224758
4  trip_distance trip_duration  0.04330031
5    fare_amount    tip_amount  0.60850233
6    fare_amount  total_amount  0.98030232
7    fare_amount trip_duration  0.22230598
8     tip_amount  total_amount  0.72666755
9     tip_amount trip_duration  0.12590691
10  total_amount trip_duration  0.21489895

Simple Linear Regression

Simple regression (e.g., fare_amount ~ trip_distance) is quick, but for demonstration, we’ll run bootstrapped regressions in parallel to estimate coefficient confidence intervals.

Code 
boot_reg <- foreach(i = 1:100, .combine = rbind) %dopar% {
  sample_idx <- sample(1:nrow(processed_chunks), replace = TRUE)
  model <- lm(fare_amount ~ trip_distance, data = processed_chunks[sample_idx, ])
  coef(model)
}

# Summarize bootstrap results
simple_reg_summary <- data.frame(
  intercept_mean = mean(boot_reg[,1]), intercept_sd = sd(boot_reg[,1]),
  slope_mean = mean(boot_reg[,2]), slope_sd = sd(boot_reg[,2])
)
simple_reg_summary
  intercept_mean intercept_sd slope_mean   slope_sd
1       18.00211    0.1503976  0.1568906 0.04376101

This provides bootstrap estimates for the intercept and slope.

Multiple Linear Regression

Similarly, for multiple regression (e.g., fare_amount ~ trip_distance + passenger_count + trip_duration), use parallel bootstrapping.

Code 
boot_multi_reg <- foreach(i = 1:100, .combine = rbind) %dopar% {
  sample_idx <- sample(1:nrow(processed_chunks), replace = TRUE)
  model <- lm(fare_amount ~ trip_distance + passenger_count + trip_duration, data = processed_chunks[sample_idx, ])
  coef(model)
}

# Summarize
multi_reg_summary <- data.frame(
  intercept_mean = mean(boot_multi_reg[,1]), intercept_sd = sd(boot_multi_reg[,1]),
  distance_slope_mean = mean(boot_multi_reg[,2]), distance_slope_sd = sd(boot_multi_reg[,2]),
  passenger_slope_mean = mean(boot_multi_reg[,3]), passenger_slope_sd = sd(boot_multi_reg[,3]),
  duration_slope_mean = mean(boot_multi_reg[,4]), duration_slope_sd = sd(boot_multi_reg[,4])
)
multi_reg_summary
  intercept_mean intercept_sd distance_slope_mean distance_slope_sd
1       16.00972    0.1857126           0.1586257        0.06161287
  passenger_slope_mean passenger_slope_sd duration_slope_mean duration_slope_sd
1            0.4737781         0.01303285          0.08431273       0.002112575

Clean Up

Stop the parallel cluster.

Code 
stopImplicitCluster()

Summary and Conclusion

The {doParallel} package is an accessible and efficient way to implement parallel computing in R, particularly for iterative tasks that can be split across multiple cores. By registering a parallel backend and using {foreach} with %dopar%, users can significantly speed up computations for data processing, statistical analyses, and simulations. This tutorial demonstrated how to leverage {doParallel} for various data processing and analysis tasks on a large dataset, showcasing its practical applications in real-world scenarios. At end of the analysis, always remember to stop the parallel cluster to free up system resources.

References

  1. doParallel CRAN Documentation
    Weston, S., & Revolution Analytics. (2022). doParallel: Foreach Parallel Adaptor. CRAN. https://cran.r-project.org/web/packages deParallel/index.html
    Details setup and usage of {doParallel} with {foreach}.

  2. Getting Started with doParallel
    Weston, S. (2019). doParallel Vignette. CRAN. https://cran.r-project.org/web/packages/doParallel/vignettes/gettingstartedParallel.pdf
    Practical guide for parallel loops with examples.

  3. Parallel R Book
    McCallum, Q. E., & Weston, S. (2011). Parallel R. O’Reilly Media.
    Covers {doParallel} for data analysis and statistical tasks.

  4. Parallel Computing in R
    Schmidberger, M., et al. (2009). Journal of Statistical Software, 31(1). doi:10.18637/jss.v031.i01
    Reviews parallel computing in R, including {parallel}.