Running Spark with QFS on a Docker Swarm

In my last post, we put together a quick build that runs Apache Spark on the Personal Compute Cluster. The goal of that build was to simply demonstrate how to run Spark on a Docker Swarm. What it did not address well was how Spark would access data. I did set up GlusterFS shared volume that did allow Spark to read and write data files. However, this was not fast because the set up did not allow Spark to take advantage of data locality when scheduling task. As a result, almost all data reads and writes are done across the ethernet backbone.

The goal in this post is to set up a Spark build on Docker Swarm that is paired with a distributed file system set up in a manner where data locality can be leveraged to speed up processing. There are choices for distributed file systems, the most notable being Hadoop’s HDFS and the Quantcast File System (QFS). For this build, I am going to use QFS. It will be my eventual goal to create a build with HDFS too, and then do a performance comparison between the two builds.

Spark Swarm Design

In order to design the Spark deployment, we first need to take an inventory of all the services required to provide a usable deployment:

One important detail is to understand is how Spark identifies data locality. When the Spark master is scheduling tasks against file parts, it considers where the QFS meta server reports the file parts are, as indicated by the chunk server’s IP address. If the Spark master knows it has a worker at the same IP address as the chunk server containing the file part, the Spark master will prefer asking that file part’s work to the Spark worker with the same IP address. This is not a guarantee, as the Spark master has a few other considerations in assigning tasks to Spark workers, but Spark worker having the same IP address as a chunk server is necessary for data locality to be leveraged.

In a Docker Swarm, there are two ways to ensure two services share the same IP address. The first is to not use private networking in the swarm cloud and instead only use the physical node’s IP address across all services running on that node. The second is to consolidate services that needs to share the same IP address into the save Docker image, thus effectively making them the same service. There are pros and cons to each approach. For my private Personal Compute Cluster (meaning, it’s mine, no one else is using it, and is not exposed to the public), the difference between the two approaches becomes more academic, and so in this build, I went with the approach that consolidates services into one Docker image. This means that a Docker image will be built that runs both the Spark worker and the QFS chunk server, and together they represent a single “worker node”. We will then run on each physical node one worker node container based on this image.

Given all that, the Docker image design is as follows:

The complete configuration and design and be reviewed in my GitHub repository.

All of the Docker containers in this build will be launched together as a Docker stack on the swarm running in the Personal Compute Cluster. There are some key details in how this stack is configured:

Initializing the Spark QFS Swarm

Prior to the first time the Spark QFS Swarm is run, we need to configure the Personal Cluster. The first task is to build the Docker images. Do that by cloning my GitHub repository and running the shell script for building the images. On the master node (if you cloned this repository from my last post, then all you should do here is a git pull):

cd repositories
git clone git@github.com:DIYBigData/personal-compute-cluster.git
cd personal-compute-cluster/spark-qfs-swarm/
./build-images.sh

Next, we need to create the various directories on the physical nodes that the various services will use. From the master node, run the following:

parallel-ssh -i -h ~/cluster/all.txt -l root "mkdir -p /mnt/data/qfs/logs"
sudo mkdir -p /mnt/data/qfs/checkpoint
parallel-ssh -i -h ~/cluster/all.txt -l root "mkdir -p /mnt/data/qfs/chunk"
parallel-ssh -i -h ~/cluster/all.txt -l root "mkdir -p /mnt/data/spark"
parallel-ssh -i -h ~/cluster/all.txt -l root "chown -R michael /mnt/data/qfs"
parallel-ssh -i -h ~/cluster/all.txt -l root "chown -R michael /mnt/data/spark"
parallel-ssh -i -h ~/cluster/all.txt -l root "chgrp -R docker /mnt/data/qfs"
parallel-ssh -i -h ~/cluster/all.txt -l root "chgrp docker /mnt/data/spark"

Of course, replace the user michael in the above commands with the user name you use on your cluster.

The next step is to initialize the QFS cluster. The initialization step creates a blank file system on the QFS cluster, and is effectively the same thing as erasing a hard drive. To initialize QFS, we will directly run the qfs-master container on the master node

docker run -it --mount type=bind,source=/mnt/data/qfs,target=/data/qfs master:5000/qfs-master:latest /bin/bash

Then when in the Docker container’s bash shell:

$QFS_HOME/bin/metaserver -c $QFS_HOME/conf/Metaserver.prp
exit

Running the Spark QFS Swarm

Once the Personal Compute Cluster has been initialized as described above, launching the Spark QFS stack is pretty straight forward. From within the spark-qfs-swarm directory of the master node clone of my git repository, simply run:

docker stack deploy -c deploy-spark-qfs-swarm.yml spark

Once everything is up and running on the swarm, simply point your development computers browser at http://your-cluster-IP-address:7777 to get the Jupyter notebook UI. While you cluster is running a spark job or Jupyter notebook, you can point your browser at http://your-cluster-IP-address:4040 to watch the progress of the Spark cluster.

This example notebook can be used to quickly test your deployment. It simply calculates all of the prime numbers up to a configurable max value, saves the resulting data frame to QFS then loads it back into another data frame, and then calculates some aggregate statistics on the found prime numbers. You can try adjusting the MAX_VALUE variable to calculate larger ranges of prime numbers.

It’s interesting to note that when I first ran this notebook, I decided to verify my results with Wolfram Alpha. I used a MAX_VALUE of 1000000000, and then asked Wolfram Alpha the count of prime number less than 1000000000. My number and their number differed. After some digging, I became convinced it was a bug in Wolfram Alpha, reported it, and they concurred. So at least in this case, my Personal Compute Cluster was able to out perform Wolfram Alpha.

In a future post I will more aggressively put this Spark deployment through some stress tests.

Shutting Down the Spark QFS Swarm

If you want to delete the Spark QFS stack running on the Docker swarm, the following command will remove it:

docker stack rm spark

To relaunch the stack later, simply launch it with the docker stack deploy command provided int eh section above.