Step 0: Get Number of Cores of the Compute Machine (my local machine, in this case)

Determine how many cores local processing machine has. This will be used to determine the number of groups to split the data into in the next set.

library(parallel)
cores <- detectCores()
cores

## [1] 4

Step 1: Add Groups

Let’s add groups to sp_500. The groups are needed to divide the data across the number cores. For me, this is 4 cores. Then create a group vector, which is a sequential vector of 1:cores (1 to 4) repeated the length of the number of rows in sp_500. Subsequently, add the group vector to the sp_500 tibble using the dplyr::bind_cols() function.

group <- rep(1:cores, length.out = nrow(sp_500))
sp_500 <- bind_cols(tibble(group), sp_500)
sp_500

Step 2: Create Clusters

Use the create_cluster() function from the multidplyr package. Think of a cluster as a work environment on a core. Therefore, the code below establishes a work environment on each of the 4 cores.

cluster <- create_cluster(cores = cores)
cluster

## socket cluster with 4 nodes on host 'localhost'

Step 3: Partition by Group

Next is partitioning. Think of partitioning as sending a subset of the initial tibble to each of the clusters. The result is a partitioned data frame (party_df), which we explore next. Use the partition() function from the multidplyr package to split the sp_500 list by group and send each group to a different cluster.

by_group <- sp_500 %>%
    partition(group, cluster = cluster)
by_group

## Source: party_df [505 x 3]
## Groups: group
## Shards: 4 [126--127 rows]
## 
## # Description: S3: party_df
##    group symbol company                        
##    <int> <chr>  <chr>                          
##  1     2 ABT    Abbott Laboratories            
##  2     2 ATVI   Activision Blizzard            
##  3     2 AES    AES Corp                       
##  4     2 APD    Air Products & Chemicals Inc   
##  5     2 ARE    Alexandria Real Estate Equities
##  6     2 AGN    Allergan, Plc                  
##  7     2 GOOGL  Alphabet Inc Class A           
##  8     2 AEE    Ameren Corp                    
##  9     2 AIG    American International Group   
## 10     2 ABC    AmerisourceBergen Corp         
## # ... with 495 more rows

The result, by_group, looks similar to our original tibble, but it is a partitioned data frame, which is very different. The key is to notice that the there are 4 Shards. Each Shard has between 126 and 128 rows, which evenly splits our data among each shard. Now that our tibble has been partitioned into a party_df, we are ready to move onto setting up the clusters.

Step 4: Setup Clusters

The clusters have a local R work environment, which doesn’t work for the vast majority of cases. Code typically depends on libraries, functions, expressions, variables, and/or data that are not available in base R. Fortunately, there is a way to add these items to the clusters.

For our computation, we are going to need to add several libraries along with the get_stock_prices() function to the clusters. This is done by using the cluster_library() and cluster_assign_value() functions, respectively.

from <- "2007-01-01"
to   <- today()

# Utilize pipe (%>%) to assign libraries, functions, and values to clusters
by_group %>%
    # Assign libraries
    cluster_library("tidyverse") %>%
    cluster_library("stringr") %>%
    cluster_library("lubridate") %>%
    cluster_library("quantmod") %>%
    # Assign values (use this to load functions or data to each core)
    cluster_assign_value("get_stock_prices", get_stock_prices) %>%
    cluster_assign_value("from", from) %>%
    cluster_assign_value("to", to)

Step 5: Run Parallelized Code

Now that we have our clusters and partitions set up and everything looks good, we can run the parallelized code. The code chunk is the same as the series code chunk with two exceptions:

1. Instead of starting with the sp_500 tibble, we start with the by_group party_df
2. We combine the results at the end using the collect() function

start <- proc.time() # Start clock

sp_500_processed_in_parallel <- by_group %>% # Use by_group party_df
    mutate(
        stock_prices = map(.x = symbol, 
                           ~ get_stock_prices(symbol = .x,
                                              return_format = "tibble",
                                              from = from,
                                              to   = to)
                           )
        ) %>%
    collect() %>% # Special collect() function to recombine partitions
    as_tibble()   # Convert to tibble

time_elapsed_parallel <- proc.time() - start # End clock

head(sp_500_processed_in_parallel)

time_elapsed_parallel

##    user  system elapsed 
##    0.46    0.15   39.75

sp_500_processed_in_parallelize <- sp_500_processed_in_parallel %>% tidyr::unnest()
#Error: Each column must either be a list of vectors or a list of data frames [issues]

#sp_500_processed_in_parallelize <- sp_500_processed_in_parallel %>% mutate(stock_prices = flatten(stock_prices))
#Error: Column `stock_prices` must be length 505 (the number of rows) or one not 3521

DT::datatable(sp_500_processed_in_parallelize, options = list(pagelength=5))

The processing time is approximately 0.38 minutes, which is 4X faster!