multidplyr

multidplyr is a backend for dplyr that partitions a data frame across multiple cores. The workflow follows these steps:

1. Partitioning: You split the data using partition(). The data stays on each worker node to minimize data movement.
2. Computation: You perform dplyr operations on the partitioned data frame. These computations are spread across multiple cores.
3. Collection: You use collect() to explicitly retrieve the processed data back to the main (host) R session.

Performance Note: Due to communication overhead, multidplyr is most effective when parallelizing slower and more complex functions. For simple operations on datasets with fewer than ~10 million observations, you may not see significant improvements and might prefer dtplyr.

You can install the released version of multidplyr from CRAN or the development version from GitHub using pak.

# From CRAN
install.packages("multidplyr")

# From GitHub (development version)
# install.packages("pak")
pak::pak("tidyverse/multidplyr")

The most efficient way to use multidplyr is to have each worker read different files directly. This avoids moving large datasets from the host to the workers.

Use cluster_assign_each() to distribute a vector of filenames, cluster_send() to execute the loading command (e.g., using vroom) on each worker, and party_df() to create the partitioned data frame object.

# 1. Create a filename vector containing different values on each worker
cluster_assign_each(cluster, filename = c("a.csv", "b.csv", "c.csv", "d.csv"))

# 2. Use vroom to quickly load the csvs on each worker
cluster_send(cluster, my_data <- vroom::vroom(filename))

# 3. Create a party_df using the my_data variable on each worker
my_data <- party_df(cluster, "my_data")

To use multidplyr, first create a cluster of workers. Each worker is a separate R process. Use new_cluster() to define the number of workers and cluster_library() to ensure necessary packages (like dplyr) are attached to each worker.

library(multidplyr)

cluster <- new_cluster(4)
cluster_library(cluster, "dplyr")

If data is already loaded in your main session, use partition() to spread it across workers.

Best Practice: Before calling partition(), call group_by() on your data frame. This ensures that all observations belonging to the same group end up on the same worker, which is essential for correct grouped computations.

library(nycflights13)
library(dplyr)

# Group by a variable to ensure group integrity on workers
flight_dest <- flights %>% 
  group_by(dest) %>% 
  partition(cluster)

After performing computations on a party_df, the results remain distributed across the workers. To bring the final result back to your main R session, use collect().

flight_dest %>% 
  summarise(delay = mean(dep_delay, na.rm = TRUE), n = n()) %>% 
  collect()