Data Processing and Analysis with {parallel}

As data sizes grow, traditional sequential data processing in R can become a bottleneck. Parallel computing allows you to leverage multiple CPU cores to speed up your data processing tasks. This tutorial will guide you through parallel data processing techniques in R with CPU cores, from basic concepts to practical applications.

The New York City Yellow Taxi Trip Data (yellow_tripdata_2023.parquet), provided by the NYC Taxi and Limousine Commission, is a rich dataset containing millions of trip records, including variables such as pickup and dropoff times, trip distances, fare amounts, and passenger counts. Analyzing such large datasets poses computational challenges, particularly for statistical tasks like descriptive statistics, correlation analysis, and regression modeling. The {parallel} package in R offers an efficient solution by distributing computations across multiple CPU cores, significantly reducing processing time for data-intensive operations.

This tutorial demonstrates how to leverage the {parallel} package to perform essential statistical analyses on the NYC taxi dataset. Specifically, we will compute descriptive statistics (e.g., mean, median, standard deviation) for key variables like trip duration, fare amount, and trip distance; calculate Pearson correlation coefficients to explore relationships between these variables; and fit both simple and multiple linear regression models to predict fare amounts based on trip duration, trip distance, and passenger count. By combining {parallel} with {arrow} for efficient Parquet file handling and {dplyr} for data manipulation, we ensure scalable and rapid analysis of large datasets. The tutorial uses parLapply() for cross-platform compatibility and mclapply() for macOS/Linux users, providing practical examples and performance comparisons to highlight the benefits of parallel processing.

Check and Install Required R Packages

Code 
packages <- c(
          'tidyverse', 
          'data.table', 
          'feather', 
          'arrow',
          'parallel',
          'foreach',
          'doParallel',
          'parallelly', 
          'nycflights13'
) 
#| 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     parallel      foreach 
        TRUE         TRUE         TRUE         TRUE         TRUE         TRUE 
  doParallel   parallelly nycflights13 
        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:nycflights13" "package:parallelly"   "package:doParallel"  
 [4] "package:iterators"    "package:foreach"      "package:parallel"    
 [7] "package:arrow"        "package:feather"      "package:data.table"  
[10] "package:lubridate"    "package:forcats"      "package:stringr"     
[13] "package:dplyr"        "package:purrr"        "package:readr"       
[16] "package:tidyr"        "package:tibble"       "package:ggplot2"     
[19] "package:tidyverse"    "package:stats"        "package:graphics"    
[22] "package:grDevices"    "package:utils"        "package:datasets"    
[25] "package:methods"      "package:base"

Overview of the {parallel} Package

The {parallel} package is a core R package that provides functions for parallel computing, allowing you to utilize multiple CPU cores to speed up computations. It supports both multi-core processing on a single machine and distributed computing across multiple machines.

Key Functions in {parallel}

  • mclapply(): A parallel version of lapply() that works on Unix-like systems (Linux, macOS). It uses forking to create child processes. Not available on Windows.

  • parLapply(): A parallel version of lapply() that works on all platforms, including Windows. It requires setting up a cluster using makeCluster().

  • makeCluster(): Creates a cluster of R processes that can run in parallel. You can specify the number of cores to use.

  • stopCluster(): Stops the cluster and frees up resources.

  • detectCores(): Detects the number of CPU cores available on the machine.

Setting Up the {parallel} Package

First, let’s check how many CPU cores are available:

Code 
# Check number of CPU cores
num_cores <- detectCores()
cat("Available CPU cores:", num_cores, "\n")
Available CPU cores: 80
Code 
# Recommended: use one less than total cores to avoid system slowdown
cores_to_use <- max(1, num_cores - 1)
cat("Recommended cores to use:", cores_to_use, "\n")
Recommended cores to use: 79

Data Processing with {parallel}

This section demonstrates parallel data processing using the NYC Yellow Taxi Trip Data (yellow_tripdata_2023.parquet). We will compute the average trip distance per passenger count using both serial and parallel approaches with the {parallel} package. The dataset is large (~3 million rows), making it suitable for parallel processing to improve performance.

Data

The dataset is in Parquet format, which can query directly without loading into memory using {arrow} pcakage. For reproducibility: - Download the January 2023 data from: https://d37ci07v2hxiua.cloudfront.net/trip-data/yellow_tripdata_2023-01.parquet (about 47 MB, ~3 million rows). - We saved it to a local folder, e.g., /home/zia207/Dropbox/WebSites/R_Website/Quarto_Projects/R_Beginner/Data/yellow_tripdata_2023-01.parquet. - We’ll also use the Taxi Zone Lookup CSV for joins: Download from https://s3.amazonaws.com/nyc-tlc/misc/taxi+_zone_lookup.csv.

The dataset includes columns like VendorID, tpep_pickup_datetime, tpep_dropoff_datetime, passenger_count, trip_distance, PULocationID, DOLocationID, fare_amount, total_amount, etc.

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

Let’s compute the average trip distance per passenger_countas an example. Since the dataset is large, we’ll split it into chunks and process them in parallel using parLapply()(cross-platform) ormclapply()` (macOS/Linux).

Splitting the Data

To parallelize, we split the dataset into chunks based on the number of cores:

Code 
# Number of chunks (equal to number of cores - 1)
n_chunks <- num_cores - 1

# Split data into chunks
data_splits <- split(taxi_data, 
                     rep(1:n_chunks, each = ceiling(nrow(taxi_data) / n_chunks), 
                         length.out = nrow(taxi_data)))

Defining the Processing Function

Define a function to compute the average trip distance per passenger_count for each chunk:

Code 
process_chunk <- function(chunk) {
  chunk %>%
    filter(!is.na(passenger_count), !is.na(trip_distance), passenger_count > 0) %>%
    group_by(passenger_count) %>%
    summarise(avg_distance = mean(trip_distance, na.rm = TRUE)) %>%
    collect()  # Ensure computation is performed
}

Serial Processing (Baseline)

First, let’s compute the result serially to establish a baseline

Code 
system.time({
  serial_result <- lapply(data_splits, process_chunk)
  serial_result <- do.call(rbind, serial_result) %>%
    group_by(passenger_count) %>%
    summarise(avg_distance = mean(avg_distance, na.rm = TRUE))
})
   user  system elapsed 
  0.569   0.068   0.636
Code 
print(serial_result)
# A tibble: 9 × 2
  passenger_count avg_distance
            <dbl>        <dbl>
1               1         3.36
2               2         3.98
3               3         3.69
4               4         3.81
5               5         3.27
6               6         3.27
7               7         4.24
8               8         4.55
9               9         0

Parallel Processing with parLapply

Now, we’ll parallelize the computation using parLapply():

Code 
# Create a cluster
cl <- makeCluster(num_cores - 1)

# Export required packages and functions to the cluster
clusterEvalQ(cl, {
  library(dplyr)
  library(arrow)
})
[[1]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[2]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[3]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[4]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[5]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[6]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[7]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[8]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[9]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[10]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[11]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[12]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[13]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[14]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[15]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[16]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[17]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[18]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[19]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[20]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[21]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[22]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[23]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[24]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[25]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[26]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[27]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[28]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[29]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[30]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[31]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[32]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[33]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[34]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[35]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[36]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[37]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[38]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[39]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[40]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[41]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[42]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[43]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[44]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[45]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[46]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[47]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[48]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[49]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[50]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[51]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[52]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[53]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[54]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[55]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[56]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[57]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[58]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[59]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[60]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[61]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[62]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[63]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[64]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[65]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[66]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[67]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[68]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[69]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[70]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[71]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[72]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[73]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[74]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[75]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[76]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[77]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[78]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[79]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"
Code 
clusterExport(cl, c("process_chunk"))

# Parallel processing
system.time({
  parallel_result <- parLapply(cl, data_splits, process_chunk)
  parallel_result <- do.call(rbind, parallel_result) %>%
    group_by(passenger_count) %>%
    summarise(avg_distance = mean(avg_distance, na.rm = TRUE))
})
   user  system elapsed 
  0.711   0.186   1.028
Code 
# Stop the cluster
stopCluster(cl)

print(parallel_result)
# A tibble: 9 × 2
  passenger_count avg_distance
            <dbl>        <dbl>
1               1         3.36
2               2         3.98
3               3         3.69
4               4         3.81
5               5         3.27
6               6         3.27
7               7         4.24
8               8         4.55
9               9         0

Parallel Processing with mclapply (macOS/Linux)

If you’re on macOS or Linux, you can use mclapply() for simpler parallelization:

Code 
system.time({
  parallel_result_mc <- mclapply(data_splits, process_chunk, mc.cores = num_cores - 1)
  parallel_result_mc <- do.call(rbind, parallel_result_mc) %>%
    group_by(passenger_count) %>%
    summarise(avg_distance = mean(avg_distance, na.rm = TRUE))
})
   user  system elapsed 
  0.037   2.568   2.631
Code 
print(parallel_result_mc)
# A tibble: 9 × 2
  passenger_count avg_distance
            <dbl>        <dbl>
1               1         3.36
2               2         3.98
3               3         3.69
4               4         3.81
5               5         3.27
6               6         3.27
7               7         4.24
8               8         4.55
9               9         0

Basic Statistical Analysis

In this section of tutorial demonstrates how to perform basic statistical analyses on the NYC Yellow Taxi Trip Data (yellow_tripdata_2023.parquet) using R’s {parallel} package to speed up computations. We’ll cover descriptive statistics, correlation analysis, simple linear regression, and multiple linear regression to analyze relationships between trip duration, fare amount, trip distance, and passenger count. The {arrow} package will handle the Parquet file, and {dplyr} will assist with data manipulation. We’ll parallelize computations to efficiently process the large dataset (~3 million rows).

We will conduct the following statistical analyses on the NYC Yellow Taxi data:

  • Descriptive Statistics: Compute mean, median, standard deviation, minimum, and maximum for trip_duration, fare_amount, and trip_distance.
  • Correlation Analysis: Calculate Pearson correlation coefficients among trip_duration, fare_amount, and trip_distance.
  • Simple Linear Regression: Estimate fare_amount based on trip_duration.
  • Multiple Linear Regression: Estimate fare_amount using trip_duration, trip_distance, and passenger_count.

We will employ the {parallel} package to parallelize computations across CPU cores, using parLapply() (a platform-independent function). and mclapply() (macOS/Linux).

Read and Preprocess Data

Code 
# Read and preprocess data
taxi_data <- read_parquet(paste0(dataFolder, "yellow_tripdata_2023.parquet")) %>%
  mutate(trip_duration = as.numeric(difftime(tpep_dropoff_datetime, tpep_pickup_datetime, units = "mins"))) %>%
  filter(
    !is.na(trip_duration), !is.na(fare_amount), !is.na(trip_distance), !is.na(passenger_count),
    trip_duration > 0, fare_amount > 0, trip_distance > 0, passenger_count > 0
  ) %>%
  select(trip_duration, fare_amount, trip_distance, passenger_count)

# Inspect the data
glimpse(taxi_data)
Rows: 35,609,468
Columns: 4
$ trip_duration   <dbl> 8.433333, 6.316667, 12.750000, 10.833333, 12.300000, 1…
$ fare_amount     <dbl> 9.3, 7.9, 14.9, 11.4, 12.8, 12.1, 45.7, 17.7, 14.9, 11…
$ trip_distance   <dbl> 0.97, 1.10, 2.51, 1.43, 1.84, 1.66, 11.70, 2.95, 3.01,…
$ passenger_count <dbl> 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 4, 1, 2, 1, 1, 1, 1, 3, …

Setting Up Parallel Processing

Code 
# Detect number of CPU cores
num_cores <- detectCores()
print(num_cores)  # E.g., 8 cores
[1] 80
Code 
# Split data into chunks
n_chunks <- num_cores - 1
data_splits <- split(taxi_data, rep(1:n_chunks, each = ceiling(nrow(taxi_data) / n_chunks), length.out = nrow(taxi_data)))

Descriptive Statistics

Compute mean, median, standard deviation, min, and max for trip_duration, fare_amount, and trip_distance.

Define the Descriptive Statistics Function

Code 
process_descriptive <- function(chunk) {
  chunk %>%
    summarise(
      mean_duration = mean(trip_duration, na.rm = TRUE),
      median_duration = median(trip_duration, na.rm = TRUE),
      sd_duration = sd(trip_duration, na.rm = TRUE),
      min_duration = min(trip_duration, na.rm = TRUE),
      max_duration = max(trip_duration, na.rm = TRUE),
      mean_fare = mean(fare_amount, na.rm = TRUE),
      median_fare = median(fare_amount, na.rm = TRUE),
      sd_fare = sd(fare_amount, na.rm = TRUE),
      min_fare = min(fare_amount, na.rm = TRUE),
      max_fare = max(fare_amount, na.rm = TRUE),
      mean_distance = mean(trip_distance, na.rm = TRUE),
      median_distance = median(trip_distance, na.rm = TRUE),
      sd_distance = sd(trip_distance, na.rm = TRUE),
      min_distance = min(trip_distance, na.rm = TRUE),
      max_distance = max(trip_distance, na.rm = TRUE)
    ) %>%
    collect()
}

Serial Descriptive Statistics

Code 
system.time({
  serial_desc <- lapply(data_splits, process_descriptive)
  serial_desc_summary <- do.call(rbind, serial_desc) %>%
    summarise(across(everything(), mean, na.rm = TRUE))
})
   user  system elapsed 
  5.123   2.848   7.977
Code 
print(serial_desc_summary)
# A tibble: 1 × 15
  mean_duration median_duration sd_duration min_duration max_duration mean_fare
          <dbl>           <dbl>       <dbl>        <dbl>        <dbl>     <dbl>
1          17.6            12.7        41.7       0.0173        2963.      19.8
# ℹ 9 more variables: median_fare <dbl>, sd_fare <dbl>, min_fare <dbl>,
#   max_fare <dbl>, mean_distance <dbl>, median_distance <dbl>,
#   sd_distance <dbl>, min_distance <dbl>, max_distance <dbl>

Parallel Descriptive Statistics with parLapply

Code 
# Create a cluster
cl <- makeCluster(num_cores - 1)

# Export required packages and functions
clusterEvalQ(cl, {
  library(dplyr)
  library(arrow)
})
[[1]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[2]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[3]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[4]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[5]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[6]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[7]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[8]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[9]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[10]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[11]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[12]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[13]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[14]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[15]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[16]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[17]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[18]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[19]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[20]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[21]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[22]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[23]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[24]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[25]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[26]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[27]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[28]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[29]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[30]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[31]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[32]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[33]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[34]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[35]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[36]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[37]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[38]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[39]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[40]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[41]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[42]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[43]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[44]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[45]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[46]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[47]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[48]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[49]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[50]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[51]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[52]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[53]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[54]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[55]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[56]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[57]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[58]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[59]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[60]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[61]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[62]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[63]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[64]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[65]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[66]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[67]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[68]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[69]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[70]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[71]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[72]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[73]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[74]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[75]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[76]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[77]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[78]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[79]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"
Code 
clusterExport(cl, c("process_descriptive"))

# Parallel descriptive statistics
system.time({
  parallel_desc <- parLapply(cl, data_splits, process_descriptive)
  parallel_desc_summary <- do.call(rbind, parallel_desc) %>%
    summarise(across(everything(), mean, na.rm = TRUE))
})
   user  system elapsed 
  1.691   0.561   3.488
Code 
# Stop the cluster
stopCluster(cl)

print(parallel_desc_summary)
# A tibble: 1 × 15
  mean_duration median_duration sd_duration min_duration max_duration mean_fare
          <dbl>           <dbl>       <dbl>        <dbl>        <dbl>     <dbl>
1          17.6            12.7        41.7       0.0173        2963.      19.8
# ℹ 9 more variables: median_fare <dbl>, sd_fare <dbl>, min_fare <dbl>,
#   max_fare <dbl>, mean_distance <dbl>, median_distance <dbl>,
#   sd_distance <dbl>, min_distance <dbl>, max_distance <dbl>

Correlation Analysis

Compute the Pearson correlation matrix for trip_duration, fare_amount, and trip_distance.

Define the Correlation Function

Code 
process_correlation <- function(chunk) {
  cor_matrix <- cor(chunk[, c("trip_duration", "fare_amount", "trip_distance")], method = "pearson", use = "complete.obs")
  as.vector(cor_matrix)
}

Serial Correlation

Code 
system.time({
  serial_cor <- lapply(data_splits, process_correlation)
  serial_cor_mean <- colMeans(do.call(rbind, serial_cor), na.rm = TRUE)
  cor_matrix <- matrix(serial_cor_mean, nrow = 3, dimnames = list(
    c("trip_duration", "fare_amount", "trip_distance"),
    c("trip_duration", "fare_amount", "trip_distance")
  ))
})
   user  system elapsed 
  1.078   0.426   1.503
Code 
print(round(cor_matrix, 3))
              trip_duration fare_amount trip_distance
trip_duration         1.000       0.263         0.071
fare_amount           0.263       1.000         0.249
trip_distance         0.071       0.249         1.000

Parallel Correlation with parLapply

Code 
cl <- makeCluster(num_cores - 1)
clusterEvalQ(cl, {
  library(dplyr)
  library(arrow)
})
[[1]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[2]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[3]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[4]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[5]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[6]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[7]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[8]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[9]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[10]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[11]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[12]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[13]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[14]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[15]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[16]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[17]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[18]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[19]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[20]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[21]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[22]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[23]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[24]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[25]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[26]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[27]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[28]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[29]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[30]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[31]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[32]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[33]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[34]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[35]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[36]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[37]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[38]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[39]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[40]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[41]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[42]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[43]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[44]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[45]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[46]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[47]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[48]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[49]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[50]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[51]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[52]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[53]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[54]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[55]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[56]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[57]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[58]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[59]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[60]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[61]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[62]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[63]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[64]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[65]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[66]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[67]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[68]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[69]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[70]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[71]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[72]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[73]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[74]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[75]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[76]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[77]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[78]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[79]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"
Code 
clusterExport(cl, c("process_correlation"))

system.time({
  parallel_cor <- parLapply(cl, data_splits, process_correlation)
  parallel_cor_mean <- colMeans(do.call(rbind, parallel_cor), na.rm = TRUE)
  parallel_cor_matrix <- matrix(parallel_cor_mean, nrow = 3, dimnames = list(
    c("trip_duration", "fare_amount", "trip_distance"),
    c("trip_duration", "fare_amount", "trip_distance")
  ))
})
   user  system elapsed 
  1.651   0.625   3.502
Code 
stopCluster(cl)

print(round(parallel_cor_matrix, 3))
              trip_duration fare_amount trip_distance
trip_duration         1.000       0.263         0.071
fare_amount           0.263       1.000         0.249
trip_distance         0.071       0.249         1.000

Parallel Correlation with mclapply (macOS/Linux)

Code 
system.time({
  parallel_cor_mc <- mclapply(data_splits, process_correlation, mc.cores = num_cores - 1)
  parallel_cor_mc_mean <- colMeans(do.call(rbind, parallel_cor_mc), na.rm = TRUE)
  parallel_cor_mc_matrix <- matrix(parallel_cor_mc_mean, nrow = 3, dimnames = list(
    c("trip_duration", "fare_amount", "trip_distance"),
    c("trip_duration", "fare_amount", "trip_distance")
  ))
})
   user  system elapsed 
  0.022  10.396  10.467
Code 
print(round(parallel_cor_mc_matrix, 3))
              trip_duration fare_amount trip_distance
trip_duration         1.000       0.263         0.071
fare_amount           0.263       1.000         0.249
trip_distance         0.071       0.249         1.000

Simple Linear Regression

Model fare_amount as a function of trip_duration

Define the Simple Regression Function

Code 
process_simple_regression <- function(chunk) {
  model <- lm(fare_amount ~ trip_duration, data = chunk)
  coef <- summary(model)$coefficients
  list(
    intercept = coef[1, 1],
    slope = coef[2, 1],
    r_squared = summary(model)$r.squared
  )
}

Serial Simple Regression

Code 
system.time({
  serial_reg <- lapply(data_splits, process_simple_regression)
  serial_reg_summary <- do.call(rbind, lapply(serial_reg, as.data.frame)) %>%
    summarise(
      avg_intercept = mean(intercept, na.rm = TRUE),
      avg_slope = mean(slope, na.rm = TRUE),
      avg_r_squared = mean(r_squared, na.rm = TRUE)
    )
})
   user  system elapsed 
 15.115   5.370  17.761
Code 
print(serial_reg_summary)
  avg_intercept avg_slope avg_r_squared
1      17.60915 0.1209458    0.07304362

Parallel Simple Regression with parLapply

Code 
cl <- makeCluster(num_cores - 1)
clusterEvalQ(cl, {
  library(dplyr)
  library(arrow)
})
[[1]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[2]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[3]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[4]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[5]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[6]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[7]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[8]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[9]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[10]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[11]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[12]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[13]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[14]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[15]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[16]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[17]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[18]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[19]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[20]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[21]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[22]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[23]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[24]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[25]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[26]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[27]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[28]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[29]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[30]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[31]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[32]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[33]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[34]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[35]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[36]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[37]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[38]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[39]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[40]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[41]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[42]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[43]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[44]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[45]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[46]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[47]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[48]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[49]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[50]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[51]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[52]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[53]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[54]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[55]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[56]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[57]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[58]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[59]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[60]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[61]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[62]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[63]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[64]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[65]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[66]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[67]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[68]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[69]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[70]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[71]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[72]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[73]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[74]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[75]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[76]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[77]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[78]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[79]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"
Code 
clusterExport(cl, c("process_simple_regression"))

system.time({
  parallel_reg <- parLapply(cl, data_splits, process_simple_regression)
  parallel_reg_summary <- do.call(rbind, lapply(parallel_reg, as.data.frame)) %>%
    summarise(
      avg_intercept = mean(intercept, na.rm = TRUE),
      avg_slope = mean(slope, na.rm = TRUE),
      avg_r_squared = mean(r_squared, na.rm = TRUE)
    )
})
   user  system elapsed 
  1.650   0.761   3.802
Code 
stopCluster(cl)

print(parallel_reg_summary)
  avg_intercept avg_slope avg_r_squared
1      17.60915 0.1209458    0.07304362

Parallel Simple Regression with mclapply (macOS/Linux)

Code 
system.time({
  parallel_reg_mc <- mclapply(data_splits, process_simple_regression, mc.cores = num_cores - 1)
  parallel_reg_mc_summary <- do.call(rbind, lapply(parallel_reg_mc, as.data.frame)) %>%
    summarise(
      avg_intercept = mean(intercept, na.rm = TRUE),
      avg_slope = mean(slope, na.rm = TRUE),
      avg_r_squared = mean(r_squared, na.rm = TRUE)
    )
})
   user  system elapsed 
 18.690  56.872  11.256
Code 
print(parallel_reg_mc_summary)
  avg_intercept avg_slope avg_r_squared
1      17.60915 0.1209458    0.07304362

Multiple Linear Regression

Model fare_amount as a function of trip_duration, trip_distance, and passenger_count

Define the Multiple Regression Function

Code 
process_multiple_regression <- function(chunk) {
  model <- lm(fare_amount ~ trip_duration + trip_distance + passenger_count, data = chunk)
  coef <- summary(model)$coefficients
  list(
    intercept = coef[1, 1],
    slope_duration = coef[2, 1],
    slope_distance = coef[3, 1],
    slope_passenger = coef[4, 1],
    r_squared = summary(model)$r.squared
  )
}

Serial Multiple Regression

Code 
system.time({
  serial_multi_reg <- lapply(data_splits, process_multiple_regression)
  serial_multi_reg_summary <- do.call(rbind, lapply(serial_multi_reg, as.data.frame)) %>%
    summarise(
      avg_intercept = mean(intercept, na.rm = TRUE),
      avg_slope_duration = mean(slope_duration, na.rm = TRUE),
      avg_slope_distance = mean(slope_distance, na.rm = TRUE),
      avg_slope_passenger = mean(slope_passenger, na.rm = TRUE),
      avg_r_squared = mean(r_squared, na.rm = TRUE)
    )
})
   user  system elapsed 
 16.171   3.241  19.415
Code 
print(serial_multi_reg_summary)
  avg_intercept avg_slope_duration avg_slope_distance avg_slope_passenger
1      15.10938          0.1046761          0.5276607           0.6830048
  avg_r_squared
1     0.1888502

Parallel Multiple Regression with parLapply

Code 
cl <- makeCluster(num_cores - 1)
clusterEvalQ(cl, {
  library(dplyr)
  library(arrow)
})
[[1]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[2]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[3]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[4]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[5]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[6]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[7]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[8]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[9]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[10]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[11]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[12]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[13]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[14]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[15]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[16]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[17]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[18]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[19]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[20]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[21]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[22]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[23]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[24]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[25]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[26]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[27]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[28]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[29]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[30]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[31]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[32]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[33]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[34]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[35]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[36]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[37]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[38]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[39]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[40]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[41]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[42]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[43]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[44]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[45]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[46]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[47]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[48]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[49]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[50]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[51]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[52]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[53]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[54]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[55]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[56]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[57]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[58]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[59]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[60]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[61]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[62]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[63]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[64]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[65]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[66]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[67]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[68]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[69]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[70]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[71]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[72]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[73]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[74]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[75]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[76]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[77]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[78]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"     

[[79]]
[1] "arrow"     "dplyr"     "stats"     "graphics"  "grDevices" "utils"    
[7] "datasets"  "methods"   "base"
Code 
clusterExport(cl, c("process_multiple_regression"))

system.time({
  parallel_multi_reg <- parLapply(cl, data_splits, process_multiple_regression)
  parallel_multi_reg_summary <- do.call(rbind, lapply(parallel_multi_reg, as.data.frame)) %>%
    summarise(
      avg_intercept = mean(intercept, na.rm = TRUE),
      avg_slope_duration = mean(slope_duration, na.rm = TRUE),
      avg_slope_distance = mean(slope_distance, na.rm = TRUE),
      avg_slope_passenger = mean(slope_passenger, na.rm = TRUE),
      avg_r_squared = mean(r_squared, na.rm = TRUE)
    )
})
   user  system elapsed 
  1.814   0.797   4.097
Code 
stopCluster(cl)

print(parallel_multi_reg_summary)
  avg_intercept avg_slope_duration avg_slope_distance avg_slope_passenger
1      15.10938          0.1046761          0.5276607           0.6830048
  avg_r_squared
1     0.1888502

Parallel Multiple Regression with mclapply (macOS/Linux)

Code 
system.time({
  parallel_multi_reg_mc <- mclapply(data_splits, process_multiple_regression, mc.cores = num_cores - 1)
  parallel_multi_reg_mc_summary <- do.call(rbind, lapply(parallel_multi_reg_mc, as.data.frame)) %>%
    summarise(
      avg_intercept = mean(intercept, na.rm = TRUE),
      avg_slope_duration = mean(slope_duration, na.rm = TRUE),
      avg_slope_distance = mean(slope_distance, na.rm = TRUE),
      avg_slope_passenger = mean(slope_passenger, na.rm = TRUE),
      avg_r_squared = mean(r_squared, na.rm = TRUE)
    )
})
   user  system elapsed 
 20.309  61.327  11.253
Code 
print(parallel_multi_reg_mc_summary)
  avg_intercept avg_slope_duration avg_slope_distance avg_slope_passenger
1      15.10938          0.1046761          0.5276607           0.6830048
  avg_r_squared
1     0.1888502

Key points to Consider When Using Parallel Processing in R

Rule of thumb: only parallelize when tasks are substantial

Code 
# Rule of thumb: only parallelize when tasks are substantial
should_parallelize <- function(task_time, setup_time = 0.1, overhead_factor = 2) {
  # Task should be at least 10x the setup time
  task_time > (setup_time * overhead_factor * 10)
}

# Example usage
task_time <- 0.5  # seconds
if(should_parallelize(task_time)) {
  cat("Consider parallel processing\n")
} else {
  cat("Sequential processing is sufficient\n")
}
Sequential processing is sufficient

Memory Management

Code 
# Monitor memory usage
get_memory_usage <- function() {
  if(Sys.info()["sysname"] == "Windows") {
    # Windows-specific memory check
    mem <- system("tasklist /FI \"IMAGENAME eq Rterm.exe\" /FO CSV /NH", 
                  intern = TRUE)
    # Parse memory from output
    return("Memory monitoring not implemented for Windows")
  } else {
    # Unix/Linux/Mac
    mem_kb <- system("ps -o rss= -p $(pgrep R)", intern = TRUE)
    return(as.numeric(mem_kb) / 1024)  # Convert to MB
  }
}

# Use garbage collection
perform_gc <- function() {
  gc(verbose = FALSE)
  cat("Garbage collection performed\n")
}

Error Handling and Debugging

Code 
# Robust parallel processing with error handling
safe_parallel_process <- function(data_list, processor_func, n_cores = NULL) {
  if(is.null(n_cores)) {
    n_cores <- max(1, detectCores() - 1)
  }
  
  # Set up cluster
  cl <- makeCluster(n_cores)
  clusterExport(cl, "processor_func")
  
  # Process with error handling
  tryCatch({
    results <- parLapply(cl, data_list, function(chunk) {
      tryCatch({
        processor_func(chunk)
      }, error = function(e) {
        list(error = TRUE, message = e$message, traceback = traceback())
      })
    })
  }, error = function(e) {
    cat("Parallel processing failed:", e$message, "\n")
    results <- NULL
  })
  
  # Clean up
  stopCluster(cl)
  
  return(results)
}

Best Practices

  • Filter early: Remove NA values and invalid data (e.g., passenger_count <= 0) before splitting to save memory.

  • Optimize chunks: Make sure chunks are balanced (similar row counts) to improve core utilization.

  • Memory management: Large datasets like NYC taxi data can use a lot of memory. Use {arrow}’s memory-efficient Parquet reading to reduce RAM usage.

  • Cluster cleanup: Always call stopCluster(cl) after using parLapply() to free resources.

  • Test serially: Debug with lapply() before switching to parallel to ensure correctness.

Limitations

  • Windows: Use parLapply() because mclapply() is not supported.

  • Overhead: Parallelization has setup costs (e.g., cluster creation, data splitting). It is only beneficial for computationally intensive tasks.

  • I/O bottlenecks: Reading the Parquet file is not parallelized here; consider parallel I/O with tools like {duckdb} for very large datasets.

parallelly: Enhancing the {parallel} Package

The parallelly package in R is an enhancement of the base R parallel package, designed to improve and extend parallel computing capabilities. Part of the Futureverse ecosystem, parallelly provides additional tools for managing parallel workers, including querying, killing, and cloning nodes, with support for both local and remote machines. It is backward compatible with the parallel package, ensuring seamless integration, and adds features like automatic detection of CPU cores, simplified remote cluster setup, and automatic worker cleanup. Key functions include availableCores(), makeClusterPSOCK(), isNodeAlive(), killNode(), and cloneNode(), which offer more robust control over parallel processes compared to the base parallel package.

Differences from the parallel Package

  1. Enhanced CPU Core Detection: parallelly’s availableCores() considers system settings, cgroups, Linux containers, and job scheduler configurations, falling back to parallel::detectCores() if needed, unlike the simpler detectCores() in parallel.

  2. Improved Remote Cluster Setup: makeClusterPSOCK() in parallelly simplifies remote worker setup without requiring manual IP address or firewall configuration, unlike parallel::makePSOCKcluster().

  3. Advanced Worker Management: parallelly introduces isNodeAlive() and killNode() for checking and terminating workers (local and remote) and cloneNode() for restarting failed nodes, features absent in parallel.

  4. Automatic Cleanup: parallelly::autoStopCluster() ensures clusters shut down automatically when garbage-collected, reducing stray processes, unlike parallel.

  5. Futureverse Integration: While parallelly supports the future package, it can be used independently, unlike parallel, which is more tightly coupled to base R.

Below is an example of using parallelly to process the NYC Taxi yellow_tripdata_2023-01.parquet dataset, which contains taxi trip data. The example reads the Parquet file, processes trip distances in parallel to calculate total distance per passenger count, and compares it to sequential processing.

Code 
# Function to compute total trip distance per passenger count
calc_distance <- function(df_chunk) {
  aggregate(trip_distance ~ passenger_count, data = df_chunk, sum)
}

# Sequential processing
system.time({
  result_seq <- calc_distance(taxi_data)
})
   user  system elapsed 
 33.306   5.804  39.115
Code 
# Inspect the data
print("Dataset summary:")
[1] "Dataset summary:"
Code 
summary(taxi_data[, c("passenger_count", "trip_distance")])
 passenger_count trip_distance      
 Min.   :1.000   Min.   :1.000e-02  
 1st Qu.:1.000   1st Qu.:1.090e+00  
 Median :1.000   Median :1.800e+00  
 Mean   :1.393   Mean   :3.637e+00  
 3rd Qu.:1.000   3rd Qu.:3.420e+00  
 Max.   :9.000   Max.   :1.617e+05
Code 
# Filter out invalid rows (e.g., NA or negative passenger_count/trip_distance)
taxi_data <- taxi_data[!is.na(taxi_data$passenger_count) & 
                       !is.na(taxi_data$trip_distance) & 
                       taxi_data$passenger_count > 0 & 
                       taxi_data$trip_distance >= 0, ]

# Check if data is non-empty after filtering
if (nrow(taxi_data) == 0) {
  stop("No valid data after filtering. Check passenger_count and trip_distance columns.")
}

# Function to compute total trip distance per passenger count
calc_distance <- function(df_chunk) {
  # Return NULL if the chunk is empty or has no valid rows
  if (nrow(df_chunk) == 0 || all(is.na(df_chunk$passenger_count)) || all(is.na(df_chunk$trip_distance))) {
    return(NULL)
  }
  # Aggregate only if valid data exists
  aggregate(trip_distance ~ passenger_count, data = df_chunk, sum, na.rm = TRUE)
}
Code 
# Sequential processing
system.time({
  result_seq <- calc_distance(taxi_data)
})
   user  system elapsed 
 33.157   5.009  38.173
Code 
# Parallel processing with parallelly
system.time({
  # Create a cluster with parallelly, limiting to available connections
  max_cores <- min(availableCores() - 1, 45) # Use at most 45 cores
  cl <- makeClusterPSOCK(max_cores)
  on.exit(parallelly::autoStopCluster(cl)) # Ensure cluster cleanup

  # Split data into chunks, ensuring non-empty chunks
  n_cores <- length(cl)
  data_chunks <- split(taxi_data, cut(seq_len(nrow(taxi_data)), n_cores, labels = FALSE))

  # Export function and required objects to cluster
  clusterExport(cl, c("calc_distance"))

  # Parallel computation
  results <- parLapply(cl, data_chunks, calc_distance)

  # Filter out NULL results and combine
  results <- results[!sapply(results, is.null)] # Remove empty results
  if (length(results) == 0) {
    stop("No valid results from parallel computation. Check data chunks.")
  }
  result_par <- do.call(rbind, results)
  result_par <- aggregate(trip_distance ~ passenger_count, data = result_par, sum, na.rm = TRUE)
})
   user  system elapsed 
  3.959   3.408  10.175
Code 
# Print results
print("Sequential Result:")
[1] "Sequential Result:"
Code 
print(result_seq)
  passenger_count trip_distance
1               1   95260733.60
2               2   22891954.98
3               3    5406489.73
4               4    3250928.79
5               5    1633565.51
6               6    1057429.06
7               7        350.83
8               8        623.39
9               9        297.20
Code 
print("Parallel Result:")
[1] "Parallel Result:"
Code 
print(result_par)
  passenger_count trip_distance
1               1   95260733.60
2               2   22891954.98
3               3    5406489.73
4               4    3250928.79
5               5    1633565.51
6               6    1057429.06
7               7        350.83
8               8        623.39
9               9        297.20

Summary and Conclusion

Parallel data processing is a critical technique for managing large-scale data and computationally intensive tasks. By distributing workloads across multiple processing units, it achieves significant speedups, scalability, and efficiency compared to sequential methods. It relies on effective data partitioning, task coordination, and robust frameworks like Spark or Hadoop. As data volumes grow and computational demands increase, parallel processing remains essential for industries and research fields requiring fast, scalable solutions.

The {parallel} package enables efficient computation of descriptive statistics, correlation, and regression analyses on large datasets like the NYC Yellow Taxi data. By splitting the data and using parLapply() or mclapply(), we analyzed trip_duration, fare_amount, trip_distance, and passenger_count with significant speedups. Combining {parallel} with {arrow} ensures scalability. For further analysis, consider grouping by other variables or exploring nonlinear models.

References

  1. R Core Team (2023). “parallel: Support for Parallel Computation in R.”
    • Description: Official documentation for the {parallel} package, included in base R. It provides detailed explanations of functions like mclapply() and parLapply(), with examples for parallelizing computations.
    • Source: Available in R via help(package = "parallel") or online at https://cran.r-project.org/web/packages/parallel/parallel.pdf
    • Relevance: Primary reference for understanding the {parallel} package’s functionality and implementation.
  2. McCallum, E., & Weston, S. (2011). “Parallel R.” O’Reilly Media.
    • Description: A practical guide to parallel computing in R, covering the {parallel} package and its predecessors (snow, multicore). It includes examples of parallelizing data analysis tasks.
    • Source: Available as a book or through O’Reilly’s online platform: https://www.oreilly.com/library/view/parallel-r/9781449309909/
    • Relevance: Offers practical insights into applying {parallel} for data analysis, including large dataset processing.
  3. Mayer, M. (2017 “¿Parallel Computing in R with the parallel Package.”
    • Description: A tutorial blog post that explains how to use the {parallel} package for tasks like data processing and statistical analysis, with code examples for parLapply() and mclapply().
    • Source: Available at https://www.r-bloggers.com/2017/03/parallel-comte-in-r/
    • Relevance: Provides beginner-friendly examples of parallelizing statistical computations, relevant to correlation and regression.
  4. NYC Taxi and Limousine Commission (2023). “TLC Trip Record Data.”
    • Description: Official source for the NYC Yellow Taxi Trip Data, including the yellow_tripdata_2023-01.parquet file. It includes metadata and documentation for the dataset used in the analysis.
    • Source: https://www.nyc.gov/site/tlc/about/tlc-trip-record-data.page
    • Relevance: Essential for understanding the structure and variables of the taxi dataset used in the statistical analyses.
  5. Wickham, H., & Grolemund, G. (2016). “R for Data Science.” O’Reilly Media.
    • Description: A comprehensive guide to data analysis in R, including sections on data manipulation with {dplyr} and handling large datasets. While not specific to {parallel}, it provides context for combining {dplyr} with parallel processing.
    • Source: Available online at https://r4ds.had.co.nz/ or as a book.
    • Relevance: Useful for understanding {dplyr} operations (e.g., filtering, grouping) used in conjunction with {parallel} for statistical analysis.