Kubernetes who?

The hype around kubernetes is real, but likely also justified. Kubernetes is an open-source tool that facilitates deployment of jobs and services onto computer clusters. It provides different patterns for different type of workloads, be it API servers, databases or running batch jobs. Not only makes kubernetes running workloads and services easy, it also keeps them running.

At the core of the technology are containers, which kubernetes skillfully manages inside so-called pods. A pod represents a single instance of an application and contains one or sometimes more containers. Pods in turn live on worker nodes - actual servers - and are managed by a controller on the master node. We can interact with pods indirectly via instructions to controller.

Mark Edmondson has already written a fantastic blog post about different use cases for running R application inside kubernetes. I’ll dive into the one topic he didn’t expand upon: the parallel execution of R code on kubernetes.

I will similarly use GCP’s kubernetes engine to deploy my jobs, however all major cloud providers have similar offerings. It’s worth mentioning that Google provides 300$ worth of credit free to spend on any of their cloud products, so you can freely experiment without burning a hole in your pocket.

Single job with static parameters

The simplest use case of parallelization is running the same script over and over again, but in parallel instead of in a sequential order. A classic example is simulation, i.e. the random generation of numbers given a fixed set of parameters.

I am taking an example from Azure’s tutorial on running R code in parallel, simulating stock prices after a year (365 days) given a fixed value for standard deviation and average stock price movement per day.

mean_change = 1.001 
volatility = 0.01 
opening_price = 100 

getClosingPrice <- function(days) { 
        movement <- rnorm(days, mean=mean_change, sd=volatility) 
        path <- cumprod(c(opening_price, movement)) 
        closingPrice <- path[days] 
        return(closingPrice) 
} 

replicate(1000, getClosingPrice(365)) 

Let’s save the above script into an executable file, in our case monte-carlo.R, and write a minimal Dockerfile encapsulating the script. Remember kubernetes works with containers and can access them directly from Dockerhub.

FROM rocker/r-base
COPY monte-carlo.R ./

We build the image and upload it to Dockerhub using the docker command line tool.

# build image
docker build -t mtoto/mc-demo:latest .
# upload to docker hub
docker push mtoto/mc-demo:latest

Now comes the kubernetes bit in the form of a job.yaml file, that contains the instructions for the controller. Note that under spec: we specify the number of pods to run our job on in parallel (distribution of pods over nodes is handled by kubernetes), and the number of completions. Each pod picks up a single run and exists after the script has finished. By the end of this workload 100 pods have been created, run and terminated.

apiVersion: batch/v1
kind: Job
metadata:
  name: static-demo
spec:
  parallelism: 10
  completions: 100
  template:
    metadata:
      name: static-example
      labels:
        jobgroup: static-example
    spec:
      containers:
      - name: birthday
        image: mtoto/mc-demo
        command: ["Rscript", "monte-carlo.R"]
      restartPolicy: Never

With everything in place (R script, Dockerfile, .yaml file), we are ready to deploy our first job to kubernetes. Assuming you have enabled the relevant services in the google cloud console, downloaded the google cloud SDK and have kubectl installed, we can create our cluster and deploy our first the workload on GCP in the following way:

# create 4 node cluster "kubepar" on google kubernetes engine
gcloud container clusters create kubepar --machine-type n1-standard-1 --num-nodes 4
# get credentials to point kubectl to our cluster
gcloud container clusters get-credentials kubepar
# create job
kubectl create -f job.yaml

We can monitor the progress of our job using the command kubectl get pods, to see how many pods have successfully run.

Similarly we can look at the state of the nodes with kubectl get nodes or the overall status of the job with kubectl get jobs static-demo. For a more detailed output, substitute get with describe, such as kubectl describe pods.

Once the job has finished, we collect the output of our simulation from the logs of each pod and write it to a .txt file.

for p in $(kubectl get pods -l jobgroup=static-example -o name)
do
  kubectl logs $p >> output.txt
done

Reading the output into R we can plot the results:

hist(stockprices)

Common template and multiple parameters using expansion

Moving on, we want to parallelize a script with different parameters at each run. Again, I am taking an example from a doAzureParallel tutorial where

… we calculate for a room of N people the probability that someone in the room shares a birthday with someone else in the room.

Below is the simulation script for 100.000 rooms where we supply the number of people in the room as a command line argument.

#!/usr/bin/env Rscript
args = commandArgs(trailingOnly=TRUE)
n <- as.double(args[1])

pbirthdaysim <- function(n) { 
        ntests <- 100000 
        pop <- 1:365 
        anydup <- function(i) 
                any(duplicated( 
                    sample(pop, n, replace=TRUE)))
        sum(sapply(seq(ntests), anydup)) / ntests 
}

pbirthdaysim(n)

Unlike before, we are not creating a single representation of our Job object in a .yaml file, but a Job template with placeholders. The Dockerfile is the same as before, except for the script. Don’t forget to build and upload the image before continuing.

apiVersion: batch/v1
kind: Job
metadata:
  name: par-demo-$ITEM
spec:
  template:
    metadata:
      name: par-example
      labels:
        jobgroup: par-example
    spec:
      containers:
      - name: birthday
        image: mtoto/birthday-demo
        command: ["Rscript", "birthday.R $ITEM"]
      restartPolicy: Never

Notice that we didn’t specify parallelization parameters nor the number of completions. It’s because we are going to expand the above template into 100 different job.yaml files, one for each run with a different n parameter ($ITEM in the .yaml fie) for the birthday simulation.

# create folder for jobs
mkdir jobs
# create job.yaml files
for i in {1..100}
do
  cat job.yaml | sed "s/\$ITEM/$i/" > ./jobs/job-$i.yaml
done

With the same command as before, we create all the jobs at once: kubectl create -f ./jobs. Kubernetes will automatically create, distribute and run our jobs in parallel across pods on the nodes of our cluster.

Using the same bash script as before, we can retrieve the output from each run and after read it into R.

Plotting the results, the probability that 2 or more people will have the same birthday is 99% after 60 people are in the room.

plot(probabiliy, xlab="People in room", 
     ylab="Probability of shared birthday")

Fine parallel processing using a work queue

In the previous example, we created all the jobs at once, which can overload the scheduler if the number of jobs is very large. A smarter approach is to create a work queue and let the pods pick them off one by one as they go along. Unlike before, each pod will work on multiple items until the queue is empty instead of creating a pod for each task.

To illustrate the last approach, we will parallelize the training of different regression models, a common use case for parallelization in R.

The function below takes the name of an algorithm, loads the dataset, creates a training set, runs a model using the caret package and finally uploads the result to google cloud storage as an .rds file. This way the work queue only needs to contain the names of the models to run.

# modeling function
run_save_model <- function(method) {
        # load pkgs and data
        library(mlbench)
        library(caret)
        data("BostonHousing")
        # split data
        set.seed(123)
        train_index <- createDataPartition(BostonHousing$medv,1, p = .7)
        train <- BostonHousing[train_index[[1]],]
        # train model
        model <- train(medv ~., 
                       data = train, 
                       method = method)
        
        # upload to storage bucket
        file <- sprintf("%s_model.rds", method)
        saveRDS(model, file)
        googleCloudStorageR::gcs_upload(file, 
                   name = file,
                   bucket = "bostonmodels")
}

# create redis pod and redis service
kubectl create -f ./redis-pod.yaml
kubectl create -f ./redis-service.yaml
# create temporary interactive pod
kubectl run temp -i --rm --tty  --image redis --command "/bin/sh"
# initiate redis cli
redis-cli -h redis
# push items into queue named "test"
rpush test "lm" "rf" "gbm" "enet" "brnn" "bridge"
# doublecheck queue
lrange test 0 -1

# create secret named "gcs-key"
kubectl create secret generic gcs-key --from-file=key.json=PATH-TO-KEY-FILE.json

source("rediswq.R")
source("functions.R")

# connect to redis host
host <- Sys.getenv("REDIS_SERVICE_HOST")
db <- redis_init(host = host)
vars_init("test")

# authenticate gcs
library(googleCloudStorageR)

print(paste0("Worker with sessionID: ", session))
print(paste0("Initial queue state: empty=", as.character(empty())))

while (!empty()) {
        item <- lease(lease_secs=10,
                        block = TRUE,
                        timeout = 2)
        if (!is.null(item)) {
                print(paste0("working on: ", item))
                # actual work
                run_save_model(item)
                complete(item)
        } else {
          print("waiting for work")       
        }
}
print("queue emtpy, finished")

apiVersion: batch/v1
kind: Job
metadata:
  name: fine-demo
spec:
  parallelism: 4
  template:
    metadata:
      name: fine-example
      labels:
        jobgroup: fine-example
    spec:
      volumes:
      - name: google-cloud-key
        secret:
          secretName: gcs-key
      containers:
      - name: c
        image: mtoto/ml-demo
        volumeMounts:
        - name: google-cloud-key
          mountPath: /var/secrets/google
        env:
        - name: GCS_AUTH_FILE
          value: /var/secrets/google/key.json
        command: ["Rscript", "worker.R"]
      restartPolicy: OnFailure

Trade-offs to keep in mind

Going from static workloads to setting up work queues that feed into the workers, we are introducing additional complexity. It’s not always a good thing, especially not if modifying existing applications is costly. We could’ve done parallel machine learning just as well using parameter expansion (the second approach).

On the other hand, having one Job object for each work item creates some overhead that a single Job object for all work items does not. Again, the difference will become more apparent the more work we have.

Lastly, the first two approaches create as many pods as work items, requiring less modification to existing code. With the last approach however each pod can process multiple items, which is a gain in efficiency.