large scale data processing at IXL and how to make it happen
Design Doc: Large-Scale Data Processing
Spark is a framework to parallelize many computations across many compute resources easily. Spark can run on top of Mesos which is a framework that makes allocating compute resources very easy. All of these frameworks can coordinate with Zookeeper to provide service discovery and high availability. This design discusses using Spark on top of Mesos coordinated through Zookeeper.
Imeh and Eduardo need to process a ton of data as quickly and efficiently as possible.
We have a lot of compute power. Peak traffic hours, most machines are pretty well utilized, but during “off hours”, many of those machines sit idle with plenty of compute time going to waste.
With our current infrastructure, adding new machines to run a single dedicated task is becoming expensive and inefficient. Especially to meet our current business needs and unknown future business needs, we must be much more flexible.
One framework, Spark, allows a developer to spread expensive computations over to worker nodes in a Spark cluster. This is mandatory if we want to reach our data processing goals within any reasonable amount of time.
However, Spark by itself is not a very flexible framework. It is possible to create a single Spark cluster, but resizing it dynamically is not a trivial task. Also, should we ever consider to run a second framework like Storm, it would require setting up a second cluster.
Apache Mesos is a generalized compute cluster framework that allows other frameworks like Spark, Storm, and many others including Hadoop to piggy back off of to allocate compute resources. Mesos is easy to scale and contract to allow including more compute resources to the Mesos cluster as machines become idle and removing them when they are not needed.
This section is better to be quoted directly from the Mesos white paper.
Mesos consists of a master process that manages slave daemons running on each cluster node, and frameworks that run tasks on these slaves.
The master implements fine-grained sharing across frameworks using resource offers. Each resource offer contains a list of free resources on multiple slaves. The master decides how many resources to offer to each framework according to a given organizational policy, such as fair sharing, or strict priority. To support a diverse set of policies, the master employs a modular architecture that makes it easy to add new allocation modules via a pluggin mechanism. To make the master fault tolerant we use ZooKeeper…to implement the failover mechanism…
A framework running on top of Mesos consists of two components: a scheduler that registers with the master to be offered resources, and an executor process that is launched on slave nodes to run the framework’s tasks. While the master determines how many resources are offered to each framework, the frameworks’ schedulers select which of the offered resources to use. When a frameworks accepts offered resources, it passes to Mesos a description of the tasks it wants to run on them. In turn, Mesos launches the tasks on the corresponding slaves.
Review Mesos architecture.
These are some of the assumption I have made when setting up Mesos.
If you want Mesos master servers to run with any sort of High Availability, Zookeeper is required. This also simplifies configuration because you can point every framework to Zookeeper to find the Mesos cluster. However, it is possible to run Mesos without Zookeeper. Mesos only operates with a single master server; there is no sharding or clustering capability within it itself. However, HA could be implemented with clever use of HA Proxy and DNS switching.
Some form of name resolution exists, either with DNS or
files. This is especially important when you wish to navigate the
Mesos web UI or use Spark. Most frameworks require a hostname or IP
that can resolve to a hostname (this includes Spark), also no hostname
can resolve to 127.0.0.1.
Choice of Distro
To build Mesos from source, you need a distribution of linux that ships with GCC that implements C++11. There are also a list of dependencies, the best example of these are listed as Ubuntu dependencies in their Dockerfile. All of these have analogues for CentOS and Fedora operating systems. Precompiled packages for Ubuntu and CentOS (which is what I used) exist from Mesosphere.
Installed Services and Software
All machines have Docker and some version of Java installed with the
JAVA_HOME either provided at boot time or installed to a standard
All machines have
iptables completely flushed and allowing all
The reason for this is that many ports are randomly assigned, especially in Spark. It can be configured to use a particular port, but then ports for the Mesos slaves are also randomly assigned for each task.
Spark will also make available jars required for a task via http to all the slaves, so this is just another port that needs to be available.
First draft at installing Mesos and configuring Spark.
To install mesos, the assumption made was that we are using a linux distrubution that has a pre-compiled package available from Mesosphere, in particular CentOS.
First set up the repository.
sudo rpm -Uvh http://repos.mesosphere.io/el/7/noarch/RPMS/mesosphere-el-repo-7-1.noarch.rpm
sudo yum install -y mesos
We can run Mesos master directly supplying command line arguments which looks something like:
mesos-master --zk=zk://ctest1:2181,ctest2:2181,ctest3:2181/mesos \ --work_dir=/var/lib/mesos --quorum=1 --log_dir=/var/log/mesos \ --port=5050
mesos, we need to tell it where to find zookeeper.
Create a file
/etc/mesos/zk with the contents:
Replace with the appropriate Zookeeper servers.
There is a second option,
/etc/mesos-master/quorum. This number must
be configured correctly to ensure the master log is replicated to
backup servers. If there are
n Mesos master servers, then
must be set to
n/2 + 1, eg if there are 3 Mesos masters, then quorum
2. The default value is
The install process is identical to that of the Master Master. We will
add one extra change to the configuration and it is setting the
containerizers flag to be
These Mesos slaves will also be running Spark jobs, so they have some
extra required software. They must have Java (or a
installed, as well as Docker.
sudo yum install -y docker java-1.7.0-openjdk-devel
Ensure Docker is running.
systemctl enable docker systemctl start docker
Then we can run Mesos slave directly with
mesos-slave --master=zk://ctest1:2181,ctest2:2181,ctest3:2181/mesos \ --containerizers=docker,mesos --log_dir=/var/log/mesos \ --work_dir=/var/run/mesos
Otherwise, it can be configured similar to the master using config
files if you installed the CentOS package and managed by
This is done by writing into the file
Now we enable and start both
systemctl enable mesos-slave systemctl start mesos-slave
Download a Spark binary tarball. There are many prebuilt tarballs, but any of them will work.i This will be required to run the Spark driver which will launch the Spark context and run tasks on the Mesos slaves.
Spark jobs must be submit locally from the view of the Spark drivers.
Inside the extracted tarball, there are two files the need to be
configured inside the
conf directory: spark-defaults.conf and
spark-defaults.conf, we need to add an option that can tell
the Mesos slaves where to find the master and where to download a
Spark binary tarball.
spark.master mesos://zk://ctest1:2181,ctest2:2181,ctest3:2181/mesos spark.executor.uri http://path/to/same-tarball-downloaded.tgz
spark-env.sh needs to export an evironment variable telling
Spark where it can find the Mesos native library.
Now this should be enough to submit a sample job to the Mesos cluster.
./bin/spark-submit \ --class org.apache.spark.examples.SparkPi lib/spark-examples*.jar 10
Spark Coarse-Grained Mode on Mesos
spark-defaults.conf file, you can set an option for
coarse-grained mode in spark. This will have Spark use all cpus
offered by mesos immediately. By default, Spark runs in fine-grained
mode, where each Spark task is a single Mesos task and resources get
used fairly but has a little overhead. This is explained more in the
Availability and Resilience
The Mesos white paper can better describe how they maintain Mesos master availability:
Since all the frameworks depends on the master, it is critical to make the master fault-tolerant. To achieve this we use two techniques. First, we have designed the master to be soft state, i.e., the master can reconstruct completely its internal state from the periodic messages it gets from the slaves, and from the framework schedulers. Second, we have implemented a hot-standby design, where the master is shadowed by several backups that are ready to take over when the master fails. Upon master failure, we use ZooKeeper 4 to select a new master from the existing backups, and direct all slaves and framework schedulers to this master. Subsequently, the new master will reconstruct the internal state from the messages it receives from slaves and framework schedulers.
Reconstructing the state depends on our choice of the value for
quorum. To be able to fully reconstruct the state,
quorum must be
larger than half the number of master servers.
The Mesos slaves are also resilient in that they’re each expendable. If a Mesos slave dies while running a task, it gets marked as “Lost” in the Mesos master and that task will eventually get rescheduled to run on another machine.
Since we are using Zookeeper to coordinate everything, it is imperative that it is running. Otherwise slaves will not know who the leading master is and so they will not get any tasks to run.
Open Questions and Problems
Mesos is very powerful and can provide ways to not only run computational frameworks like Spark and Hadoop, but can run general tasks and even services inside Docker containers. This means you can use Mesos to coordinate running services like Cassandra, Redis, and even web application servers like Tomcat.
There is also a framework called Chronos which essentially lets you run distributed cron jobs.
Spark Rejecting Resources
Spark will reject offers from Mesos if the number of CPUs a Spark task
requires (by default
1) is less than half the available CPUs on a
This is something that was not clearly documented, but should not be an issue if using Mesos slaves that have at least 2 CPUs. Just good to keep in mind.
Still an open problem. For especially long jobs, slave recovery from accidental mesos termination may become something very important. There is a plan for Slave Recovery in Mesos and can be configured. However, a Mesos slave can only recover if the executor and tasks were continuing to run. A server reboot, for example, would not be recovered.
If running services over Mesos, one problem is handling NAT or any other networking. Some frameworks exist to help tackle this like Flannel as part of CoreOS, also there is Kubernetes and in the near future Mesophere DCOS.
Will Mesos be around?
Despite there having been an explosion of frameworks to run on top of Mesos, another problem is finding which one will persist and be used one year from now, two years from now, forever?
To use an unprivileged (non-
root) user, eg
mesos, there are a few
considerations that need to be made. An open problem is determining
the best method of handling this situation. There was a point where if
the Spark driver, unconfigured for users, after submitting a job to
Mesos, would see all the jobs fail because they were attempting to run
the jobs under the same user as the one who submitted the job
rbruns. Not every machine had this user, so those jobs
So through out most of this testing, I was certain the
existed on every machine and that was what was used. Using another
user we could just need to ensure all machines have the same user and
all Spark jobs are submitted for that user.
Because many tasks will be running on randomly assigned servers, Mesos provides a rudimentary method of viewing logs, but a more robust system of remote logging would be preferred.
Currently, Spark cannot reference any files inside a Spark job. If a Spark job references a local file, every Mesos Slave must have the same file at the same path specified in the Spark job. This is not very practical, so using some sort of shared storage is preferred.
There are an abundance of ways to accomplish this though the preferred methods seem to be HDFS or S3. Luckily, there is a way to easily set up HDFS over Mesos.