R/development/snippet-parallel_script.R

In bfatemi/ninjar: Data Science Utility Belt

library(parallel)
library(tcltk)

##
## DEFINE PARALLEL FUNCTION AND INPUT VECTOR
##  - input to be split and sent as args to parallel
##    executions of function provided here.
##
input_vec <- ${0:INPUT}
para_func <- function(${1:ARG}){
  ${3:BODY}
}

##
## PARALLELIZATION PARAMETERS
##  1. Set num cores to set cluster and split input
##  2. Split to distribute accross cluster
##
ncores <- parallel::detectCores()

input_list <- lapply(
  parallel::splitIndices(length(input_vec), ncores),
  function(i) input_vec[i])

##
## PROGRESS BARS (WIN ONLY) PARAMS
##    - set iter var and get total iters for each node
##
iter <- 0
nmax <- ceiling(length(input_vec)/ncores)


##
## INITIALIZE CLUSTER
##
cl <- parallel::makeCluster(ncores)

# Export necessary tracking variables so all clusters can access them
parallel::clusterExport(cl, varlist = list("nmax", "iter", "para_func"))

# Load necessary libraries and start progress bar on each cluster
parallel::clusterEvalQ(cl, {
  library(tcltk)

  pb_title <- "PB TITLE"
  pb_label <- "PB LABEL"

  # Define iter variable and progress bar that will be updated on each cluster
  pb <- tcltk::tkProgressBar(title = pb_title,
                             label = pb_label,
                             initial = iter,
                             max = nmax)
  invisible(NULL)
})

##
## Run parallelized code
##
clusterApplyLB(cl, input_list, function(i){
  iter <<- iter + 1
  tcltk::setTkProgressBar(pb, iter, pb_title, label = pb_label)
  return(para_func(i))
})
stopCluster(cl)