# Improving Spark Application Throughput Via Memory Aware Task Co-location: A
Mixture of Experts Approach

Vicent Sanz Marco, Ben Taylor, Barry Porter, Zheng Wang  MetaLab, School of
Computing and Communications, Lancaster University, U. K. v.sanzmarco,
b.d.taylor, b.f.porter<EMAIL_ADDRESS>
(2017; 2017)

###### Abstract.

Data analytic applications built upon big data processing frameworks such as
Apache Spark are an important class of applications. Many of these
applications are not latency-sensitive and thus can run as batch jobs in data
centers. By running multiple applications on a computing host, task co-
location can significantly improve the server utilization and system
throughput. However, effective task co-location is a non-trivial task, as it
requires an understanding of the computing resource requirement of the co-
running applications, in order to determine what tasks, and how many of them,
can be co-located. State-of-the-art co-location schemes either require the
user to supply the resource demands which are often far beyond what is needed;
or use a one-size-fits-all function to estimate the requirement, which,
unfortunately, is unlikely to capture the diverse behaviors of applications.

In this paper, we present a mixture-of-experts approach to model the memory
behavior of Spark applications. We achieve this by learning, off-line, a range
of specialized memory models on a range of typical applications; we then
determine at runtime which of the memory models, or experts, best describes
the memory behavior of the target application. We show that by accurately
estimating the resource level that is needed, a co-location scheme can
effectively determine how many applications can be co-located on the same host
to improve the system throughput, by taking into consideration the memory and
CPU requirements of co-running application tasks. Our technique is applied to
a set of representative data analytic applications built upon the Apache Spark
framework. We evaluated our approach for system throughput and average
normalized turnaround time on a multi-core cluster. Our approach achieves over
83.9% of the performance delivered using an ideal memory predictor. We obtain,
on average, 8.69x improvement on system throughput and a 49% reduction on
turnaround time over executing application tasks in isolation, which
translates to a 1.28x and 1.68x improvement over a state-of-the-art co-
location scheme for system throughput and turnaround time respectively.

Resource Modeling, Memory Management, Predictive Modeling, Task Scheduling,

††copyright: acmcopyright††doi: xx.xxx/xxx_x††isbn: xxx-xxxx-xx-
xxx/xx/xx††conference: ; ; ††journalyear: 2017††price: xx.00††ccs: Theory of
computation Distributed algorithms††ccs: General and reference
Performance††ccs: Information systems Data analytics††journalyear:
2017††copyright: acmlicensed††conference: Middleware ’17; December 11-15,
2017; Las Vegas, NV, USA††price: 15.00††doi: 10.1145/3135974.3135984††isbn:
978-1-4503-4720-4/17/12

##  1\. Introduction

Big data applications built upon frameworks such as Hive (Thusoo et al.,
2009), Hadoop (Shvachko et al., 2010) and Spark (Zaharia et al., 2010) are
commonplace. Unlike interactive jobs, many of the data analytic applications
are not latency-sensitive. Therefore, they often run as batch jobs in a data
center. However, how to effectively schedule such applications to improve the
server utilization and the system throughput remains a challenge.

Specifically, if an application task is given the entirety of main memory on
each host to which it is deployed, it is effectively preventing the host
machine from being used for any other application until the current one has
finished, even if the task does not use all of the memory. Because many data
analytic tasks do not use 100% of the CPU during execution (Awan et al., 2015;
Jiang et al., 2014) there is a significant portion of unused processing
capacity. An alternate approach is to share the computing host between
multiple application tasks (where each task does not use all of the memory),
this could save time and energy by co-locating processes more effectively on
fewer machines.

Effective task co-locations require knowledge of the application’s resource
demand. For in-memory data processing frameworks like Apache Spark, RAM
consumption is a major concern (Li et al., 2015). It is particularly important
to understand the memory behavior of the application. If we co-locate too many
applications or give too much data to a single task, such that their total
memory consumption exceeds the physical memory of the host, we could cause
memory paging onto the hard disk, or an “out-of-memory” error, slowing down
the overall system. To achieve this we need a technique to predict the precise
memory requirement of any given Spark application.

Existing task co-location schemes require either: the user to provide
information of the resource requirement (Hindman et al., 2011), or employ an
analytical (Grandl et al., 2014) or statistical model (Luk et al., 2009; Grewe
et al., 2013; Delimitrou and Kozyrakis, 2014) to estimate the resource
requirement based on historical jobs or runtime profiling. These approaches,
however, have significant drawbacks. Firstly, it is difficult for a user to
give a precise estimation of the application’s requirement; and thus, the
supplied information is often over-conservative, asking far more resources
than the application needs. Secondly, a one-size-fits-all function is unlikely
to precisely capture behaviors of diverse applications, and no matter how
parameterized the model is, it is highly unlikely that a model developed today
will always be suited for tomorrow.

In this paper, we present a generic framework to model the memory behavior of
Spark applications. As a departure from prior work that uses a fixed utility
function to model the resource requirement, we use multiple linear and non-
linear functions to model the memory requirements of various applications. We
then build a machine learning classifier to select which function should be
used for a given application and dataset at runtime. As the program
implementation, workload and underlying hardware changes, different models
will be dynamically selected at runtime. Such an approach is known as
_mixture-of-experts_ (Jacobs et al., 1991). The central idea is that instead
of using a single monolithic model, we use multiple models (_experts_) where
each expert is specialized for modeling a subset of applications. Using this
approach, each memory model is used only for the applications for which its
predictions are effective. One of the advantages of our approach is that new
functions can easily be added and are selected only when appropriate. This
means that the system can evolve over time to target a wider range of
applications, by simply inserting new functions. The result is a new way of
using machine learning for system optimization, with a generalized framework
for a diverse set of applications.

We evaluate our approach on a 40-node multi-core cluster using 44 Spark
applications that cover a wide range of application domains. We show that the
accurate memory-footprint prediction given by our approach enables the runtime
scheduler to make better use of spare computing resources to improve the
overall system throughput via task co-location. We use two distinct metrics to
quantify our results: _system throughput_ and _average normalized turnaround
time_ , and compare our approach against a state-of-art resource scheduler
(Delimitrou and Kozyrakis, 2014). Experimental results show that our approach
is highly accurate in predicting the application’s memory requirement, with an
average error of 5%. By better utilizing the memory resources of a host, our
system achieves 8.69x improvement of system throughput and a 49% reduction in
application turnaround time. This translates to a 1.28x and 1.68x improvement
over the state-of-art respectively on throughput and turnaround time.

This paper makes the following contributions:

  * •

We present a novel machine learning based approach to automatically learn how
to model the memory behavior of Spark applications (Section 3);

  * •

Our work is the first to employ mixture-of-experts for resource demand
modeling. Our generic framework allows new models to be easily added to target
a wider range of applications and performance metrics;

  * •

We show how to combine this resource modeling framework with runtime task co-
location policies to improve system throughput for Spark applications (Section
4);

  * •

Our system is immediately deployable on real systems and does not require any
modification to the application source code.

##  2\. Background and Overview

###  2.1. Apache Spark

Apache Spark is a general-purpose cluster computing framework (Zaharia et al.,
2010). with APIs in Java, Scala and Python and libraries for streaming, graph
processing and machine learning (Zaharia et al., 2010). It is one of the most
active open source projects for big data processing, with over 2,000
contributors in 2016.

Each Spark application runs as an independent set of _executor_ processes,
each with dedicated memory space for executing parallel jobs within the
application. The executors are coordinated by the _driver_ program running on
a _coordinating node_. Input data of Spark applications is stored in a shared
filesystem and organized as _resilient distributed datasets_ (RDDs) – a
collection of objects that can be operated on in parallel. Each Spark executor
allocates its own heap memory space for caching RDDs. This work exploits the
data parallel property of RDDs to characterize the application’s memory
behavior without wasting computing cycles.

###  2.2. Problem Scope

Our goal is to develop a framework to accurately predict the resource
requirement of Spark applications for arbitrary inputs. In this work, we focus
on the memory requirement as RAM resources are a major concern for in-memory
data processing frameworks like Apache Spark (Li et al., 2015). To demonstrate
the usefulness of our approach, we apply it to perform task co-location for
batched, data-analytic Spark applications. We do not consider latency-
sensitive applications, such as search, as their stringent response time
targets often require isolated execution (Lo et al., 2015).

Our approach estimates the memory footprint of a Spark executor for a given
input dataset. It then uses this information to determine if there are enough
spare resources (i.e. memory and CPU) to co-locate tasks; if there is, it
calculates how many tasks could be co-located and how much work should be
given to each task. We exploit the fact that many big data applications do not
spend all of their time at 100% CPU (Jiang et al., 2014) (in our case, the
averaged CPU load is under 40% – see Section 6.7). This observation suggests
that there are opportunities to co-locate Spark tasks without significantly
increasing the CPU contention and slow down the performance of co-running
applications (see also Section 6.8). Our approach is applied to a simple task
co-location policy in this work, yet the resulted scheme outperforms the
state-of-the-art task scheduling scheme. We want to stress that our framework
can be used by other scheduling policies to provide an estimation of the
application’s resource demand to support decision making.

Our current implementation is restricted to applications whose memory
footprint is a function of their input size, this is a typical behavior for
many data analytical applications. In this work, we do not explicitly model
disk and network I/O contention, because prior research suggests that they
have little impact on the performance on the type of the applications we
target (Ousterhout et al., 2015). It is to note that this observation may not
hold for I/O intensive applications such as database workloads (Rupprecht et
al., 2017). We are also aware that not all applications’ memory consumption is
correlated to the input size and would require adding new functions to make
predictions based on other parameters, such as the model size of a machine
learning algorithm. Nonetheless, our framework is general and allows new
models to be easily added to target different applications, or other
performance and resource metrics in the future.

###  2.3. Overview of Our Approach

Our approach, depicted in Figure 1, is _completely automated_ , and no
modification to the application source code is required.

Our mixture-of-experts framework for memory footprint prediction consists of a
range of distinct models built off-line. An expert selector decides which
model should be invoked, based on the runtime information of the application.
To use our resource modeling framework to perform task co-location, a task
scheduler follows a number of steps described as follows.

For each “ _new_ ” application that is ready to run, we predict which of the
_off-line_ learned experts, _termed ‘memory function’ in this paper_ , best
describes its memory behavior, i.e. how the memory footprint changes as the
input size varies. The selection of the memory function is based on runtime
information of the program, such as the number of L1 data and instruction
cache misses. This information is collected by running the application on a
small portion (around 100MB) of the input data items111We choose this modest
input size as an input of this size typically takes a short time to process,
while at the same time, it is sufficiently large (i.e. this often results in a
working set that is larger than the size of the L3 data cache in most of the
high-end CPUs) to capture the cache behavior of the application..

Figure 1. Overview of our approach. For an incoming application, our approach
first extracts the features of the program. Based on the feature values, it
predicts which of the off-line learned memory functions best describes the
memory behavior of the application. It then instantiates the function
parameters by profiling the application on some small sets of the input data
items. A runtime scheduler then utilizes the memory function to perform task
co-location.

We then calibrate the selected function to tailor its parameters to the target
program and input. We do so by first profiling the application with two small
different-sized parts of the application input to instantiate two function
parameters; we then use the measured memory footprints to instantiate the
parameter values. The calibrated memory function is then used to determine how
many unprocessed data items should be allocated to an executor under a given
memory budget. During the profiling run, we also record the average CPU usage
of the application. After determining which memory function to use and
obtaining the CPU usage of the application, the runtime scheduler can spawn
new executors to run on servers that have spare memory, and if the aggregate
CPU load of co-running tasks will go over 100% (i.e. to avoid CPU contention).

Since runtime information collection and model calibration are all performed
on some unprocessed data items and contribute to the final output, no
computing cycle is wasted on profiling. Furthermore, we will re-run an
executor process in isolation if it fails because of an “out-of-memory” error,
but this was not observed in our experiments.

The key to our approach is choosing the right memory function and then using
lightweight profiling to instantiate the function parameters. An alternative
is to use extensive profiling runs at runtime to find a model to fit the
application’s memory behavior. However, doing so will incur significant
overhead and could outweigh the benefit (see Section 6.5).

In the next section, we will describe how supervised machine learning
(Alpaydin, 2010) can be used to construct the memory functions (experts) and
the expert selector to choose which function to use for any “ _unseen_ ”
applications.

##  3\. Predictive Modeling

Our approach involves using multiple memory functions (experts) to capture the
memory requirement of an application for a specific runtime input. The set of
memory functions are constructed offline on a set of example programs, and
then an expert selector dynamically chooses the best expert to use at runtime.

Our expert selector for determining the memory function is a K-nearest
neighbour (KNN)222We have also explored several alternative classification
techniques, including decision trees and neural networks. This is discussed in
Section 6.9. classifier (Keller and Gray, 1985). The input to the classifier
is a set of runtime features. Its output is a label to the memory function
that describes the memory behavior of the target application and the specific
dataset.

###  3.1. Learning Memory Functions

Figure 2. The training process. Table 1. Memory functions used in this work Modeling Technique | Formula
---|---
(Piecewise) Linear Regression | $y=m*x^{b}$
Exponential Regression | $y=m*(1-e^{(-b*x)})$
Napierian Logarithmic Regression | $y=m+ln(x)*b$

(a) Sort (b) PageRank

Figure 3.  The observed and predicted memory footprints for Sort and PageRank
from HiBench. The memory footprint of the two applications can be accurately
described using one of the memory functions listed in Table 1.

Our memory functions and expert selector are trained _off-line_ using a set of
training programs. The learned expert selector can then be used to predict
which memory function to use for any _new, unseen_ application. Figure 2
depicts the process of collecting training data to learn the memory functions
to build the expert selector. For each training program, we find a
mathematical function to model the application’s memory footprint and collect
its feature values.

During the training process, we run selected training programs in isolation on
a computing host. We profiled each training application with different sized
inputs. For each program input, we record the memory footprint of the Spark
executor process. Next, we try different mathematical modeling techniques to
discover which model best describes the relationship between input size and
memory allocation, that is, as the input size increases, how does the memory
allocation change. In this training phase, we record the memory function used
to describe each training program. Our intuition is that the memory behavior
for programs with similar characteristics will be similar. This hypothesis is
confirmed in Section 6.9.

We use a set of linear and non-linear regression techniques to model the
application’s memory behavior. Table 1 gives the full list of modeling
techniques we used in this work. Each of our models has two parameters, $m$
and $b$, to be instantiated during runtime model calibration. Here $x$ and $y$
are the input size (i.e. the number of RDD objects in our case) and the
predicted memory footprint respectively. It is worth mentioning that all the
memory functions are automatically learned from training data, treating the
applications as black boxes; new applications would similarly be learned
automatically, potentially causing the addition of new memory functions.

#### Example.

Figure 3 shows the observed memory footprint and the prediction given by our
memory function for Sort and PageRank. For these two applications, the memory
functions used in this work can accurately model their memory behaviors.
Specifically, the memory footprint, $y$, of Sort and PageRank for a given
input size, $x$, can be precisely described using an exponential function ,
$y=m*(1-e^{(-b*x)})$, where $m=5.768$, $b=4.479$ and a Napierian logarithmic
function, $y=m+ln(x)*b$, where $m=16.333$, $b=1.79$ respectively.

After building the memory functions, we need to have a mechanism to decide
which of the functions to use. One of the key aspects in building a successful
expert predictor is finding the right features to characterize the input
application task. This process of feature selection is described in the next
section. This is followed by sections describing training data generation and
then how to use the expert selector at runtime.

Table 2. Raw features, sorted by their importance Abbr. | Desc. | Abbr. | Desc. | Abbr. | Desc. | Abbr. | Desc.
---|---|---|---|---|---|---|---
L1_TCM | L1 total cache miss rate | L1_DCM | L1 data cache miss rate | vcache | % of memory used as cache | L1_STM | L1 cache store miss rate
bo | # blocks sent (/s) | L2_TCM | L2 data cache miss rate | L3_TCM | L2 total cache miss rate | cs | # context switches / s
FLOPs | # floating point operations /s | in | # interrupts / s | L2_DCM | L3 cache total miss rate | L2_LDM | L2 cache load miss rate
L1_ICM | L1 instr. cache miss rate | swpd | % of virtual memory used | L2_STM | L2 cache store miss rate | IPC | instruction per cycle
L1_LDM | L1 cache load miss rate | L2_ICM | L2 instr. cache miss rate | ID | % of idle time | WA | % of time on IO waiting
US | % spent on user time | SY | % spent on kernel time |  |  |  |

###  3.2. Runtime Features

#### Raw Features.

Expert selection is based on runtime characteristics of the application task.
These characteristics, called _features_ , are collected using system-wide
profiling tools: vmstat, Linux perf and performance counter tool PAPI.
Collected feature values are encoded to a vector of real values. We considered
22 raw features in this work, which are given in Table 2. Some of these
features are selected based on our intuition, while others are chosen based on
prior work (Yang et al., 2015). All these features can be automatically and
externally observed, without needing access to the source code.

#### Feature Scaling.

Supervised learning typically works better if the feature values lie in a
certain range. Therefore, we scaled the value for each of our features between
the range of 0 and 1. We record the maximum and minimum value of each feature
found at the training phase, and use these values to scale features extracted
from a new application during runtime deployment.

#### Feature Reduction.

Given the relatively small number of training applications, we need to find a
compact set of features in order to build an effective predictor. Feature
reduction is automatically performed through applying Principal Component
Analysis (PCA) on the scaled raw features. This technique removes the
redundant features by linearly aggregating features that are highly
correlated. After application of PCA, we use the top 5 principal components
(PCs) which account for 95% of the variance of the original feature space. We
record the PCA transformation matrix and use it to transform the raw features
of the target application to PCs during runtime deployment. Figure 4a
illustrates how much feature variance that each component accounts for. This
figure shows that prediction can accurately draw upon a subset of aggregated
feature values.

#### Feature Analysis.

To understand the usefulness of each raw feature, we apply the Varimax
rotation (Manly, 2004) to the PCA space. This technique quantifies the
contribution of each feature to each PC. Figure 4b shows the top 5 dominant
features based on their contributions to the PCs. Cache features, L1_TCM,
L1_DCM and L1_STM, are found to be important for describing memory behaviors.
This is not supervising as cache hit/miss rates are shown to be useful in
characterizing the application behavior in prior works (Cavazos et al., 2007;
Singh et al., 2009). Other features of virtual memory usage (vcache), I/O (bo)
and thread contention (cs) are also considered to be useful, but are less
important compared to cache features. Using this technique, we sort the raw
features listed in Table 2 according to the importance. The advantage of our
feature selection process is that it automatically determines what features
are useful when targeting a new computing environment where the importance of
features may change. Later in Section 6.9, we quantify the similarity of
programs mapped to the same memory function, which provides additional
evidences to justify the choice of features.

Recently, deep neural networks, such as Long short-term memory (LSTM), are
shown to be powerful in extracting features from program source code (Cummins
et al., 2017a). Since our current implementation only relies on runtime
information, techniques like (Cummins et al., 2017a) could be used to provide
additional features obtained from the source code. It is to note that to train
an effective deep neural network will require a significantly larger number of
training examples than we used in this work.

(a) Principal components (b) Most important raw features

Figure 4. The percentage of principal components (PCs) to the overall feature
variance (a), and contributions of the 5 most important raw features in the
PCA space (b).

###  3.3. Collect Training Data

We use _cross-validation_ to construct memory functions and the KNN classifer
to select which function to use. This standard evaluation technique works by
picking some target programs for testing and using the remaining ones for
training. In this work, we use benchmarks from the HiBench (Huang et al.,
2011) and BigDataBench (Gao et al., 2013) suites to build the memory models.
Later we show that our approach works well on benchmarks from the Spark-Perf
(Databricks, 2016) and the Spark-Bench (Li et al., 2015) suites, although we
did not directly train our models on them. The process of collecting training
data is described in Figure 2. To collect training data, we first extract the
feature values of each training program by running a single executor process
in isolation, using inputs with an average size of 100MB. Next, we run each
training program with different sized inputs (ranging from $\sim$300MB to
$\sim$1TB) and record the observed memory footprints. We then find a memory
function to closely fit the curve. For each training program, we record its
principal component values and the memory function. Since training is only
performed once, it is a _one-off_ cost.

Like any supervised learning based approaches, the effectiveness of our
approach relies having a sufficient volume of high-quality training data. We
find the set of training benchmarks gives good performance in this work,
because the benchmarks already cover a wide range of typical Spark
applications. However, we remark that when moving to a new application domain,
additional benchmarks may need in order to have an adequate sampling over the
program space. In this case, one will need to add more training programs or
using an automatic benchmark generator (Cummins et al., 2017b) to
automatically synthesize these programs.

###  3.4. Modeling Other Metrics and Program Phases

We believe our approach can be extended to model other metrics, e.g. CPU
contention. This involves finding appropriate raw-features, modelling
techniques for the experts and the expert selector, and employing a multi-
objective scheduling policy like (Llull et al., 2017). Besides these, the rest
of our approach for automatic feature selection and model generation remains
the same. Furthermore, while not explored in this work, our approach can model
changing program phases by e.g. treating a long-running phase as an individual
application.

##  4\. Runtime Deployment

Once we have learned the memory functions as described above, we can use a KNN
algorithm to choose an appropriate function to estimate the memory footprint
for any _unseen_ applications with a given input, and to use the prediction to
co-locate Spark executor tasks at runtime.

Figure 5. Our system predicts the memory function for each application and
monitors the memory resources of computing nodes. The runtime scheduler
creates new executors to run on computing nodes that have spare memory, and
uses the memory function to determine how many data items should be given to
the executor under a memory budget.

Our runtime system built upon YARN (Vavilapalli et al., 2013), a task and
resource manager for Spark. The co-location scheme will be triggered when more
than one Spark application is waiting to be scheduled. Figure 5 illustrates
the architecture of our system. For each application task, we predict its
memory function for its input dataset, and then use the memory function to co-
locate Spark executor processes whenever possible.

###  4.1. Memory Requirement Prediction

To determine the memory function for an application task, a runtime system
follows two steps, described as follows.

#### Memory Function Prediction.

We run the incoming application on a small set of the input RDD objects (with
an aggregated size of around 100MB) to collect and normalize feature values,
and to perform the PCA transformation. We then calculate the Euclidean
distance between the transformed input program feature vector and the feature
vector of each training program to find out the _nearest neighbor_ , i.e. the
training program that is closest to the input program in the feature space
(see also Section 6.9). We use the memory function of the nearest neighbor as
the prediction. One advantage of using KNN is that the Euclidean distance used
for closeness evaluation can be used to measure the prediction confidence,
which essentially provides a degree of soundness guarantee. For example, if an
application is too far from any training program, we can fall back to the
default scheme to run the application, and simultaneously re-train a new
memory function for future. Our current implementation performs feature
extraction by running the application on the lightly-loaded coordinating node
(where the driver program runs). The results generated in the feature
extraction phase will contribute to the final output of the application.

#### CPU Load.

We also record the average CPU usage during the profiling run, and use this
information later to determine whether co-location will cause CPU contention
among co-running tasks.

#### Model Calibration.

After we have determined the memory function, we need to instantiate the
function coefficients (i.e. $m$ and $b$ in Table 1). We calculate these by
running the application on two sets of unprocessed input data items, where the
first and the second sets contain 5% and 10% of the input items, respectively.
To determine the function parameters, we measured the memory footprints during
profiling runs, and use them together with the corresponding input sizes (i.e.
the number of data objects) to solve the memory function equation. At this
stage, we are only concerned with the application’s memory footprint but not
runtime. Therefore, profiling runs can be performed by either grouping
different application tasks to run on a single host or running the target
application with other latency-insensitive tasks. Again, the results generated
during this phase will contribute to the final output of the application.
Moreover, since the input and output data of the Spark application typically
stored in a shared filesystem, we do not need to explicitly move the data in
or out from the profiling host.

###  4.2. Resource Monitor

Each computing node runs a daemon that periodically reports to the resource
monitor its memory usage and CPU load. Our current implementation reports the
average memory usage and system load within a 5-minute window. The information
is retrieved from the Linux “/proc” system. Since this is performed at a
coarse-grained level (i.e. minutes), the overhead of monitoring and
communication is negligible. With this monitoring scheme in place, a task
scheduler can respond to execution phase changes and load variations, avoiding
over-subscribing the computing resources.

###  4.3. Job Dispatcher

By default, we use the dynamic allocation scheme of Spark to determine how
many free server nodes to use to run an application. However, the Spark
dynamic scheme is not perfect, so we utilize spare memory to spawn
_additional_ executors to run on servers that have spare resources. Also,
instead of waiting for the servers to become completely free, our approach
starts executing waiting applications as soon as possible, reducing the
turnaround time.

Once we have the memory function of the highest-priority application, the job
dispatcher will spawn a new executor for the application to run on severs that
have spare memory and if the aggregate CPU load of all co-running tasks will
not go over 100%. The dispatcher uses the memory function to determine how
much memory is needed for the remaining input (to allow us to co-locate more
applications if possible), and how many data items can be cached by the
executor under a given memory budget. To estimate the aggregate CPU load, we
add up the CPU load of the computing host (which is reported by the resource
monitor) and the average CPU usage of the application to be scheduled (which
is obtained during the profiling run for feature collection). Furthermore, the
number of data items to give to the co-located executor is dynamically
adjusted over time, adapting to the changes of execution stages and memory
resources. A naive alternative is to statically set the executor heap size to
the size of free memory. But doing so can over-subscribe the memory resources
than necessary and precludes co-locating more than two applications (see
Section 6.2).

To minimize the potential thread contention, we dynamically adjust the number
of threads (tasks) created by each executor to evenly distribute processor
cores across currently-running executors on a single host. Furthermore, to
enforce a certain degree of fairness, it is important to make sure that the
new co-running task does not use the resources that are deemed to be essential
for the currently running application. While fairness is not a focus of this
work, our prediction framework helps the scheduler in this endeavor.

##  5\. Experimental Setup

###  5.1. Platform and Benchmarks

#### Hardware.

We use a multi-core cluster with 40 nodes, each has an 8-core Xeon E5-2650 CPU
@ 2.6GHz (16 threads with hyper-threading), 64GB of DDR4 RAM, and 16GB of
swap. Nodes have SSD storage and are connected through 10Gbps Ethernet,
precluding disk and network contention.

#### Software.

Each computing node runs CentOS 7.2 with Linux kernel 3.12. We rely on the
local OS to schedule processes and do not bind tasks to specific cores. We use
Apache Spark 2.1.0 with Hadoop Yarn 2.6 as the cluster manager and HDFS as the
Spark file management system. We use the Oracle Java runtime, Java SE 8u. We
run Spark in the cluster mode. We also use the dynamic resource allocation
scheme, so that memory will be given back to Spark when an application task
completes. We run the Spark driver on a dedicated coordinating node and try to
run multiple Spark executors on a single host to improve the system
throughput. Finally, we use the Spark default configuration for memory
management, but we dynamically adjust the the number cores and heap size per
executor to match the available hardware resources.

#### Workloads.

We used 44 Java-based Spark applications from four widely used suites: HiBench
(Huang et al., 2011), BigDataBench (Gao et al., 2013), Spark-Perf (Databricks,
2016) and Spark-Bench (Li et al., 2015). We used the native Spark
implementations from these suites. These benchmarks implement the core
algorithms used in real-life applications e.g. machine learning, image and
natural language processing, and web analysis.

###  5.2. Evaluation Methodology

Table 3. Application task mixes used in the experiments Label | #App. | Label | #App. | Label | #App. | Label | #App.
---|---|---|---|---|---|---|---
L1 | 2 | L2 | 6 | L3 | 7 | L4 | 9
L5 | 11 | L6 | 13 | L7 | 19 | L8 | 23
L9 | 26 | L10 | 30 |  |  |  |

#### Runtime Scenarios.

We evaluated our scheme using ten runtime scenarios with a mix of 2 to 30
randomly selected applications, detailed in Table 3. For each scenario, we try
$\sim$100 different application mixes and make sure all benchmarks are
included in each scenario. The input size ranges from small ($\sim$300MB) and
medium ($\sim$30GB) to large ($\sim$1TB). Inputs were generated using the
input generation tool provided by each benchmark suite. In the experiments,
all tasks are scheduled on a first come first serve basis, but we stress that
our technique can be applied to _any scheduling policy_.

#### Predictive Model Evaluation.

Our memory functions (experts) and expert selector are trained using 16
benchmarks from HiBench and BigDataBench. We then apply the trained models to
all 44 benchmarks from the four benchmark suites. When there are benchmarks
from HiBench and BigDataBench present in the task group, we use _leave-one-
out-cross-validation_ to exclude the target applications from the training
program set and use the remaining benchmarks from HiBench and BigDataBench to
build our model. To provide a fair comparison, when testing an application
from one benchmark suite that has an equivalent implementation in the other
suite, we also exclude the benchmark from other suite from the training set.
For example, when testing Sort from HiBench, we exclude Sort from BigDataBench
from training.

#### Performance Report.

For each test case, we report _the geometric mean_ performance across all
configurations. We replay the schedule decisions for each test case multiple
times, until the difference between the upper and lower confidence bounds
under a 95% confidence interval setting is smaller than 5%. Furthermore, the
time spent on feature extraction, model calibration, and prediction is
included in our results.

###  5.3. Evaluation Metrics

We use two standard evaluation metrics for multi-programmed workloads: _system
throughput_ and _turnaround time_. We use the definitions given in (Eyerman
and Eeckhout, 2010), defined as follows.

#### 1\. System throughput (STP)

is a _higher is better metric_. It describes the aggregated progress of all
jobs under co-location execution over running each job one by one using
isolated execution. This is calculated as:

(1) $STP=\sum_{i=1}^{n}\frac{C_{i}^{is}}{C_{i}^{cl}}$

where _n_ is the number of application tasks to be scheduled, and $C_{i}^{is}$
and $C_{i}^{cl}$ are the execution time for task $i$ under the isolated
execution mode (is) where the task uses all available memory; and the co-
locating mode (cl) where there may be multiple tasks running on the same host.

#### 2\. Average normalized turnaround time (ANTT)

is a _smaller is better_ metric. It quantifies the time between a task being
created and its completion, indicating the average user-perceived delay. This
metric is defined as:

(2) $ANTT=\frac{1}{n}\sum_{i=1}^{n}\frac{C_{i}^{cl}}{C_{i}^{is}}$

###  5.4. Comparative Approaches

####  Quasar.

This is a state-of-the-art co-location scheme (Delimitrou and Kozyrakis,
2014). Quasar uses classification techniques to determine the characteristics
of the application to perform resource allocation, and task assignment and co-
location. Similar to our dynamic scheme, Quasar monitors workload performance
to adjust resource allocation and assignment when needed. Unlike our approach,
Quasar uses a single model for resource estimation. To provide a fair
comparison, we have implemented the Quasar classification scheme using the
same set of training programs that we used to build our models.

####  Pairwise.

This pairwise co-location scheme looks for servers with spare memory to co-
locate an additional task on the host. It sets the maximum heap size of the
co-locating task to the size of free memory, and relies on the Spark default
scheduler to determine how many RDD data items to be allocated to the co-
running task. This represents the default resource allocation policy used by
many co-location schemes (Llull et al., 2017).

####  Oracle.

We also compare our approach to the performance of an ideal predictor (Oracle)
that gives the perfect memory prediction for an application. This comparison
indicates how close our approach is to the theoretically perfect solution. The
prediction given by the Oracle scheme is obtained through profiling the
application on a given set of input RDD data items, but the profiling overhead
is not included in the results since we assume the Oracle predictor has the
ability to make prophetic prediction. Using the Oracle predictor, the runtime
scheduler can then search for the optimal number of data items to be given to
a co-running task.

(a) STP

(b) ANTT

Figure 6. Our approach outperforms Pairwise and Quasar on STP (a) and ANTT
(b). The baseline is running the applications one by one using isolated
execution. The min-max bars show the range of performance achieved across task
mixes for each runtime scenario.

|  |
---|---|---
(a) Pairwise (b) Quasar (c) Our approach

Figure 7. CPU utilization across servers when scheduling 30 Spark applications
(L10). The right-most non-zero point indicates the time when all applications
finish. Our approach leads to the highest server utilization and quickest
turnaround time.

##  6\. Experimental Results

Table 4. Application mix for the experiment shown in Figures 7 and 8. Order | App. | In.Size | Order | App. | In.Size | Order | App. | In.Size | Order | App. | In.Size | Order | App. | In.Size
---|---|---|---|---|---|---|---|---|---|---|---|---|---|---
1 | BDB.Wordcount | 30GB | 7 | HB.Scan | 30GB | 13 | SP.DecisionTree | 30GB | 19 | BDB.Kmeans | 30GB | 25 | SP.B.MatrixMult | 1TB
2 | SP.Kmeans | 1TB | 8 | HB.TeraSort | 1TB | 14 | SP.Spearman | 1TB | 20 | HB.Sort | 1TB | 26 | BDB.Sort | 30GB
3 | SP.glm-classification | 1TB | 9 | SB.Hive | 1TB | 15 | SB.MatrixFact | 1TB | 21 | SP.CoreRDD | 300MB | 27 | SB.RDDRelation | 1TB
4 | SP.glm-regression | 1TB | 10 | SP.NaiveBayes | 1TB | 16 | BDB.Grep | 1TB | 22 | SP.Gmm | 1TB | 28 | SP.Pearson | 1TB
5 | SP.Pca | 30GB | 11 | BDB.PageRank | 1TB | 17 | SB.LogRegre | 1TB | 23 | HB.Join | 1TB | 29 | SP.Chi-sq | 30GB
6 | SB.SVD++ | 1TB | 12 | HB.PageRank | 30GB | 18 | BDB.NaivesBayes | 30GB | 24 | SP.Sum.Statis | 30GB | 30 | HB.Kmeans | 1TB

In this section, unless stated otherwise, we report each approach’s
performance on STP and ANTT, by normalizing the results to a _baseline_ that
schedules the applications one by one with each application exclusively using
all the memory of each allocated computing node. The normalized STP and ANTT
are referred to as _normalized STP_ and _ANTT reduction_ (shown in percentage)
respectively.

###  6.1. Highlights

The highlights of our evaluation are as follows:

  * •

With the help of our mixture-of-experts approach, a simple task co-location
scheme achieves, on average, a 8.69x improvement on STP and a 49% reduction on
ANTT over isolated execution. This translates to a 1.28x and 1.68x improvement
on STP and ANTT respectively, when compared to Quasar. See Section 6.2;

  * •

Our approach is highly accurate in predicting the memory footprint of Spark
applications, with an error of less than 5% for most cases. See Section 6.9;

  * •

Our scheme is low-overhead. The time spent on feature extraction and model
calibration is less than 10% of the total application execution time, and the
profiling runs contribute to the final results. See Section 6.6;

  * •

We thoroughly evaluate our scheme by comparing it against several alternative
task co-location schemes and modeling techniques, and performing a detailed
analysis on the working mechanism of the approach.

###  6.2. Overall Performance

#### STP.

Figure 6 (a) confirms that task co-location improves system throughput. As the
number of tasks to be scheduled increases, we see an overall increase in the
STP. Pairwise performs reasonably well for small task groups, but it misses
significant opportunities for large task groups. For L9 and L10, Pairwise only
delivers half of the Oracle performance. This is because Pairwise does not
scale up beyond pairwise co-location. Quasar performs significantly better
than Pairwise by using a classifier model to coordinate resources among co-
locating tasks, but it is not as good as our approach. By employing multiple
functions to model diverse applications, our approach constantly outperforms
Pairwise and Quasar across all task groups. For large task groups (L8 - L10),
our approach delivers over 1.72x and 1.48x improvement on the STP over
Pairwise and Quasar respectively. Overall, Quasar gives on average 6.6x
improvement on STP, which translates to 65.7% of the Oracle performance. Our
approach achieves 8.69x improvement on STP, which translates to a 1.28x
improvement over Quasar or 83.9% of the Oracle performance.

#### ANTT.

Figure 6 (b) shows the ANTT reduction over the baseline. By maximizing the
system throughput, task co-location in general leads to favorable ANTT
results, particularly for large task groups. Quasar and our approach
outperforms Pairwise on ANTT by a factor of over 4x from L2 onward. Our
approach delivers better turnaround time over Quasar, by avoiding memory
contention among co-locating Spark tasks. On average, our approach reduces the
turnaround time by 49% across different task groups. This translates to 93.4%
of the Oracle performance. When compared with the 54% Oracle performance given
by Quasar, our approach achieves 1.68x better turnaround time.

#### Summary.

We achieve 83.9% and 93.4% of the Oracle performance for STP and ANTT
respectively, outperforming Pairwise, a widely used co-location policy, and
Quasar, a state-of-the-art co-location policy. The advantage of our approach
is largely attributed to its use of multiple models instead of just one to
precisely capture an applications’ memory behavior. Without this accurate
information, the alternative scheme often over- or under-provisions resources,
leading to worse performance.

(a) STP (higher is better) (b) Turnaround Time (lower is better)

Figure 8. Resultant STP (a) and turnaround time (b) for the scheduling
scenario in Figure 7. Our approach gives better STP and faster turnaround time
when compared with alternative co-location schemes.

###  6.3. Server Utilization

Figure 7 shows the CPU utilization across 40 computing nodes for Pairwise,
Quasar and our approach when scheduling 30 Spark applications (L10) using
different input sizes. Table 4 gives the application mix for this experiment,
while Figure 8 presents the turnaround time (i.e. the wall clock time to
finish the set of jobs) given by each approach. Additionally, Table 4 shows
the applications and the input size performed that where used to create Figure
7. By carefully co-locating tasks using memory footprint predictions, our
approach gives the best server utilization, which in turn leads to the highest
STP (1.81x and 1.39x higher STP over Pairwise and Quasar respectively)
quickest turnaround time (1.46x and 1.28x faster turnaround time over Pairwise
and Quasar respectively).

(a) STP

(b) ANTT

Figure 9. Compare to unified model based approaches that use a single modeling
technique to describe the application’s memory behavior.

###  6.4. Compare to Unified Models

Figure 9 compares our scheme to approaches that use one modeling technique to
predict the application’s memory footprint. In addition to the three memory
functions listed in Table 1, we also compare our scheme to a 3-layer
artificial neural network (ANN) trained using a backpropagation algorithm. We
use the same training data to build the ANN model to predict the memory
footprint. The input to the ANN model is the same set of features used by our
approach. Among the single model approaches, the ANN gives the best
performance due to its ability to model linear and non-linear behaviors. Our
approach outperforms ANN and all other approaches on STP and ANTT. The results
suggest the need for using multiple modeling techniques to capture the diverse
application behaviors. This work develops a generic framework to support this.

###  6.5. Compare to Online Search

(a) STP

(b) ANTT

Figure 10. Compare to using online search to allocate input for a given memory
budget. Our approach significantly outperforms the online search scheme,
because it avoids the runtime overhead associated with finding the optimal
number of data items to be given to the co-running task.

Figure 10 compares our approach to a method that uses descent gradient search
to dynamically adjust the right input size for a given memory budget. The
online search based approach gives rather disappointing results due to the
large overhead involved in finding the right input size. Furthermore, this
approach also suffers from a scalability issue, i.e. the searching overhead
grows as the number of computing nodes increases. Our approach avoids the
overhead by directly predicting the memory footprint, leading to 2.4x and 2.6x
better performance on STP and ANTT respectively.

###  6.6. Profiling Overhead

Figure 11. Average profiling time to total task execution time. It is to note
that during feature extraction and model calibration, the application always
executes a portion of the unprocessed data, no computing cycles are wasted.
Figure 12. Average profiling time to total runtime per program for HiBench and
BigDataBench.

The stack chart in Figure 11 shows the average time spent on feature
extraction and model calibration with respect to the total execution time per
evaluation scenario. Figure 12 gives a breakdown on per benchmark basis using
an input size of around 280GB. As profiling is performed on a single host
(thus having little communication overhead) using small inputs, the cost is
moderate. Overall, the time spent on feature extraction and model calibration
accounts for 5% and 8% respectively to the total execution time. We stress
that profiling runs also contribute to the final output of the task, so no
computing cycles are wasted; and profiling is performed while the application
is waiting to be scheduled.

###  6.7. CPU Load in Isolation Mode

Figure 13 shows the average CPU load when a benchmark is running in isolation
using all the system’s memory exclusively. The CPU load for most of the 44
benchmarks is under 40%. As a result, the CPU is often not fully utilized when
just running on application. This is in line with the finding reported by
other researchers (Awan et al., 2015). Our approach exploits this
characteristic to improve the system throughput through task co-location.

Figure 13. CPU load distributions across benchmarks when the application is
executed in the isolation mode.

###  6.8. Co-location Interferences

Figure 14. Violin plot showing the distribution of slowdown when using our
scheme to co-locate the target benchmark with another application on a single
host. The baseline is running the target application in isolation. Here we run
each of the 16 target benchmarks from HiBench and BigDataBench along with each
of the remaining 43 benchmarks.

#### Interferences among Spark Benchmarks.

The violin plot in Figure 14 shows the distribution of slowdown when running
each of the 16 benchmarks from HiBench and BigDataBench along with each of the
remaining 43 benchmarks using our scheme. The shape of the violin corresponds
to the slowdown distribution. The thick black line shows where 50% of the data
lies. The white dot is the position of the median. In the experiment, we first
launch the target application and then use the spare memory to co-locate
another competing workload. The input size of the target program is
$\sim$280GB. As can be seen from the figure, the slowdown across applications
is less than 25% and is less than 10% on average. For applications with little
computation demand, such as HB.Sort, the slowdown is minor (less than 5%). For
benchmarks with higher computation demand, such as HB.Aggregation, we observe
greater slowdown due to competing of computing resources among co-locating
tasks. Overall, our co-location scheme has little impact on the application’s
performance.

#### Interferences to PARSEC Applications.

We further extend our experiments to investigate the impact for co-locating
Spark tasks with other computation-intensive applications. For this purpose,
we run some computation-intensive C/C++ applications from the PARSEC benchmark
suite (v3.0) (Bienia, 2011) using the large, native input provided by the
suite. Figure 15 shows the slowdown distribution of each PARSEC benchmark when
they run together with each of the 44 Spark benchmarks under our scheme. As
all PARSEC benchmarks are share-memory programs, this experiment was conducted
on a single host. As expected, we observe some slowdown to the computation-
intensive PARSEC benchmark, but the slowdown is modest – less than 30%. For
most of cases, the slowdown is less than 20%. Given the significant benefit on
system throughput and server utilization given by our approach, we argue that
such a small slowdown is acceptable when maximizing the server utilization is
desired (which is typical for many data center applications). There are other
schemes such as Bubble-Flux (Yang et al., 2013) for reducing the interference
via dynamically pausing non-critical tasks, which are orthogonal to our
scheme.

Figure 15.  The slowdown distribution of computation-intensive PARSEC
benchmarks when they run with a Spark task under our scheme. The baseline is
running the target application in isolation.

###  6.9. Model Analysis

Figure 16. Program feature space. The original feature space is projected into
2 dimensions using PCA. Programs can be grouped into three clusters and mapped
to the three memory models described in Table 1.

#### Program Distribution.

Figure 16 visually depicts the distribution of benchmarks on the feature
space. To aid clarity, we use PCA to project the dimension of the original
feature space down to two. Each point in the figure is one of the 44
benchmarks. This diagram clearly shows that the 44 benchmarks can be grouped
into three clusters. After inspecting each cluster, we found that we indeed
use the same memory function (given on the figure) for all benchmarks in a
cluster. This diagram justifies the chosen number of memory functions. It also
confirms our assumption that programs with similar features can be modeled
using similar memory functions. To measure the similarity of programs within
each cluster, we calculate the Pearson correlation coefficient of each program
to its cluster center on the 2-d feature space shown in Figure 16. Our results
show that the correlation coefficient is above 0.9999 for all program, with
most programs have a correlation coefficient of 1.0 (the strongest
correlation). This confirm that our features are effective in capturing the
similarity of programs that use the same memory function.

We want to highlight that one of the advantages of our KNN classifier is that
the distance used to choose the _nearest neighbor_ program gives a confidence
estimation of how good the predicted memory function will be. If the target
application is far from any of the clusters in the feature space, it suggests
that a new memory modeling technique will be required (and our approach allows
new memory functions to be easily inserted), or a conservative co-location
policy should be used to avoid saturating the memory system.

Figure 17. Predicted memory footprints vs measured values for HiBench (HB) and
BigDataBench (BDB).

#### Prediction Accuracy.

Figure 17 compares the predicted optimal memory allocation against the
measured value, using an input size of around 280GB. All the models are
trained and evaluated using “leave-one-out-cross-validation” (see also Section
5.2)). The prediction error of our approach is less than 5% in most cases
except for HB.PageRank, BDB.PageRank and BDB.Sort for which our approach over-
provisions around 8% to 12% of the memory. This translates to 1.5GB to 2GB of
memory. Our approach also slightly under-estimates the memory requirement for
some benchmarks, but the difference is small so it does not significantly
affect the performance. In general, the accuracy can be improved by using more
training programs and more sophisticated modeling techniques to better capture
the application memory requirement, which is our future work. In practical
terms, one can also slightly over-provision (e.g. 10%) the memory allocation
to applications with higher priorities to tolerate potential prediction
errors. Overall, our approach can accurately predict the optimal memory
allocation, with an average prediction error of 5%.

|  |  |
---|---|---|---
|  |  |

(a) HB.TeraSort, HB.Wordcount (b) HB.Sort, BDB.Grep

(c) HB.Aggre, HB.Scan (d) HB.Joins, BDB.Wordcount

(e) BDB.Con.Comp, BDB.Kmeans (f) BDB.Sort, BDB.PageRank

(g) HB.Bayes, HB.Kmeans (h) BDB.N.Bayes, HB.PageRank

Figure 18. Comparisons of the predicted memory footprint to the measured
value. This set of memory functions can precisely capture the memory
requirement of our benchmarks.

#### Compare to Alternative Classifiers.

Table 5 gives the memory function prediction accuracy (averaged across
benchmarks and inputs) of various alternative classification techniques and
our KNN model. The alternative models were built using the same features and
training data. Thanks to the high-quality features, all classifiers are highly
accurate in predicting the memory function. We choose KNN because its accuracy
is comparable to alternative techniques but does not require re-training when
a new memory function is added.

#### Memory Functions.

Figure 18 compares the predicted memory footprint to the measured values for
HiBench and BigDataBench, showing that our memory functions can precisely
capture the application’s memory footprint. This figure also shows that a
single model is unlikely to capture diverse application behaviors. We address
this by developing an extensible framework into which we can easily plug-in
multiple models to capture different application behaviors.

Table 5. Prediction accuracy for different classifiers Classifier | Accuracy (%) | Classifier | Accuracy (%)
---|---|---|---
Naive Bayes | 92.5 | SVM | 95.4
MLP | 94.1 | Random Forests | 95.5
Decision Tree | 96.8 | ANN | 96.9
KNN | 97.4 |  |

##  7\. Related Work

Our work lies at the intersection between big data workload tuning and machine
learning based system optimization. There is no existing work which is similar
to ours, in respect to co-locating big data applications with optimal memory
allocation using predictive modeling.

###  7.1. Optimizing Big Data Workloads

#### Domain-specific Optimization.

There exists a large body of work focusing on optimizing a single application
using domain-specific knowledge. Prior work in domain-specific optimizations
for single big data applications includes query optimization (Binnig et al.,
2013; Condie et al., 2010; Xu et al., 2015), graph or data flow optimization
(Fokoue et al., 2016; Kyrola et al., 2012; Borkar et al., 2011; Sparks et al.,
2013), task tuning (Cheng et al., 2014) and personal assistant and deep
learning services (Hauswald et al., 2015b). By contrast, we target resource
modeling of Spark applications and demonstrate that this technique is useful
for scheduling multiple application tasks.

#### Memory Management.

Numerous techniques have been proposed to manage memory resources of big data
applications (Tang et al., 2011). Many of the prior works require using
dedicated APIs to rewrite the application (Nguyen et al., 2015; Fang et al.,
2015). Fang _et al._ introduce Interruptible Tasks, a parallel data task that
can be interrupted upon memory pressure. Their work aims to solve the out-of-
memory problem when processing large amounts of data on a single host (Fang et
al., 2015). This work is thus orthogonal to our work and can be used to
address the problem of occasional over-subscription of memory resources.
MemTune is a recent work on heap management for Spark applications (Xu et al.,
2016). It detects memory contention and dynamically adjusts the memory
partitions between Spark processes, but it does not address the problem of
precise memory allocation.

#### Application Scheduling.

Verma _et al._ use profiling information to schedule jobs within a MapReduce
application (Verma et al., 2012). Mashayekhy _et al._ develop energy-aware
heuristics to map tasks of a big data application to servers to minimize
energy usage (Mashayekhy et al., 2015). Unlike our work, all these works
target scheduling jobs within a single application, and allocate all physical
memory of a machine to one single application. Other work looks at mapping
parallelism by determining the number of cores and process time to be
allocated to an application (Hauswald et al., 2015a). Our method promotes
memory utilization on a local host, allowing the system to perform more tasks
than previously allowed with current methods. Consequently, higher multi-
tasking levels may lead to an increase in non-local data accesses within each
task; the scheduling framework in (Hauswald et al., 2015a) is therefore
complementary to our work.

#### Task Co-location.

Prior studies in task co-location include Bubble-Flux (Yang et al., 2013),
Quasar (Delimitrou and Kozyrakis, 2014), Tetris (Grandl et al., 2014) and
Cooper (Llull et al., 2017), which co-locate tasks across machines. Other
studies schedule workloads on multi-core processors (Zhuravlev et al., 2010;
Liu and Li, 2014). All the approaches mentioned above employ a single
monolithic function to model the resource requirement of application tasks.
There is little ability to examine whether the function fits the application
under the current runtime scenario. Other fine-grained scheduling frameworks,
like Mesos (Hindman et al., 2011), rely on the user to provide the resource
requirement of the application (Hindman et al., 2011). By contrast, we develop
an extensive framework that uses multiple modeling techniques to automatically
estimate the resource requirement. Our approach allows new models to be added
over time to target a wider range of applications. Experimental results show
that our approach yields better performance than a single model based
approach. On the other hand, the co-location policies developed in these prior
works for determining which two applications should co-locate are
complementary to our work.

###  7.2. Predictive Modeling

Recent studies have shown that machine learning based predictive modeling is
effective in code optimization (Tournavitis et al., 2009; Stock et al., 2012;
Simon et al., 2013; Ganapathi et al., 2009; Wang et al., 2014; Ogilvie et al.,
2017), parallelism mapping (Wang and O’Boyle, 2009, 2010; Wang and O’boyle,
2013; Wang et al., 2015; Taylor et al., 2017), task scheduling (Grewe et al.,
2013; Ren et al., 2017), and processor resource allocation (Wen et al., 2014).
In (Emani and Boyle, 2015), a mixture-of-experts approach is proposed to
schedule OpenMP programs on multi-cores. Their approach uses multiple linear
regression models to predict the optimal number of threads to use for a given
program on a single machine. Our approach differs from (Emani and Boyle, 2015)
in two aspects. First, we target a different problem (determining the memory
footprint vs the number of threads) and a different scale (multiple vs a
single node). Secondly, we use different modeling techniques, both linear and
non-linear, to capture the memory behaviors of different applications. No work
so far has used predictive modeling to model an application’s memory
requirement to co-locate big data application tasks. This work is the first to
do so.

##  8\. Conclusions

This paper has presented a novel scheme based on a mixture-of-experts approach
to estimate the memory footprint of a Spark applications for a given dataset.
Our approach determines at runtime, which of the off-line learned functions
should be used to model the application’s memory resource demand. One of the
advantages of our approach is that it provides a mechanism to gracefully add
additional expertise knowledge to target a wider range of applications. We
combine our resource prediction framework with a runtime task scheduler to co-
locate latency-insensitive Spark applications, aiming to improve system
throughput and application turnaround time. Using the accurate prediction
given by our framework, a runtime task scheduler can efficiently dispatch
multiple applications to run concurrently on a single host to improve the
system’s throughput and at the same time to ensure the total memory
consumption does not exceed the physical memory of the host. Our approach is
applied to 44 representative big data applications built upon Apache Spark. On
a 40-node cluster, our approach achieves, on average, 83.9% and 93.4% of the
Oracle performance on system throughput and turnaround time, respectively.

## Acknowledgement

We thank the anonymous reviewers for their insightful feedback. This work was
partly supported by the UK EPSRC under grants EP/M01567X/1 (SANDeRs),
EP/M029603/1 (Deep Online Cognition) and EP/M015793/1 (DIVIDEND). The
corresponding author of this paper is Zheng Wang (Email:
z.wang@lancaster.ac.uk).

## References

  * (1)
  * Alpaydin (2010) Ethem Alpaydin. 2010\.  Introduction to Machine Learning (2nd ed.).  The MIT Press.
  * Awan et al. (2015) Ahsan Javed Awan, Mats Brorsson, Vladimir Vlassov, and Eduard Ayguade. 2015.  Performance characterization of in-memory data analytics on a modern cloud server. In Big Data and Cloud Computing (BDCloud), 2015 IEEE Fifth International Conference on. 1–8.
  * Bienia (2011) Christian Bienia. 2011\.  Benchmarking Modern Multiprocessors.  Ph.D. Dissertation. Princeton University.
  * Binnig et al. (2013) Carsten Binnig, Norman May, and Tobias Mindnich. 2013\.  SQLScript: Efficiently analyzing big enterprise data in SAP HANA. In Lecture Notes in Informatics (LNI), Proceedings - Series of the Gesellschaft fur Informatik (GI), Vol. P-214. 363–382.
  * Borkar et al. (2011) Vinayak Borkar, Michael Carey, Raman Grover, Nicola Onose, and Rares Vernica. 2011.  Hyracks: A flexible and extensible foundation for data-intensive computing. In Data Engineering (ICDE), 2011 IEEE 27th International Conference on. 1151–1162.
  * Cavazos et al. (2007) John Cavazos, Grigori Fursin, Felix Agakov, Edwin Bonilla, Michael FP O’Boyle, and Olivier Temam. 2007\.  Rapidly selecting good compiler optimizations using performance counters. In Proceedings of the International Symposium on Code Generation and Optimization. IEEE Computer Society, 185–197.
  * Cheng et al. (2014) Dazhao Cheng, Jia Rao, Yanfei Guo, and Xiaobo Zhou. 2014\.  Improving MapReduce Performance in Heterogeneous Environments with Adaptive Task Tuning. In Proceedings of the 15th International Middleware Conference (Middleware ’14). ACM, New York, NY, USA, 97–108.
  * Condie et al. (2010) Tyson Condie, Neil Conway, Peter Alvaro, Joseph M. Hellerstein, Khaled Elmeleegy, and Russell Sears. 2010.  MapReduce Online. In Proceedings of the 7th USENIX Conference on Networked Systems Design and Implementation (NSDI’10).
  * Cummins et al. (2017a) Chris Cummins, Pavlos Petoumenos, Zheng Wang, and Hugh Leather. 2017a.  End-to-end Deep Learning of Optimization Heuristics. In The 26th International Conference on Parallel Architectures and Compilation Techniques (PACT).
  * Cummins et al. (2017b) Chris Cummins, Pavlos Petoumenos, Zheng Wang, and Hugh Leather. 2017b.  Synthesizing benchmarks for predictive modeling. In Code Generation and Optimization (CGO), 2017 IEEE/ACM International Symposium on. 86–99.
  * Databricks (2016) Databricks. 2016\.  Spark-Perf.  (2016).  https://github.com/databricks/spark-perf
  * Delimitrou and Kozyrakis (2014) Christina Delimitrou and Christos Kozyrakis. 2014.  Quasar: Resource-efficient and QoS-aware Cluster Management. In Proceedings of the 19th International Conference on Architectural Support for Programming Languages and Operating Systems (ASPLOS ’14).
  * Emani and Boyle (2015) Murali Krishna Emani and Michael Boyle. 2015.  Celebrating Diversity: A Mixture of Experts Approach for Runtime Mapping in Dynamic Environments.  SIGPLAN Not. 50, 6, 499–508.
  * Eyerman and Eeckhout (2010) Stijn Eyerman and Lieven Eeckhout. 2010.  Probabilistic Job Symbiosis Modeling for SMT Processor Scheduling.  SIGPLAN Not. 45, 3, 91–102.
  * Fang et al. (2015) Lu Fang, Khanh Nguyen, Guoqing Xu, Brian Demsky, and Shan Lu. 2015.  Interruptible Tasks: Treating Memory Pressure As Interrupts for Highly Scalable Data-parallel Programs. In Proceedings of the 25th Symposium on Operating Systems Principles (SOSP ’15).
  * Fokoue et al. (2016) Achille Fokoue, Oktie Hassanzadeh, Mohammad Sadoghi, and Ping Zhang. 2016.  Predicting Drug-Drug Interactions Through Similarity-Based Link Prediction Over Web Data. In Proceedings of the 25th International Conference Companion on World Wide Web (WWW ’16 Companion). International World Wide Web Conferences Steering Committee, Republic and Canton of Geneva, Switzerland, 175–178.
  * Ganapathi et al. (2009) Archana Ganapathi, Kaushik Datta, Armando Fox, and David Patterson. 2009.  A Case for Machine Learning to Optimize Multicore Performance. In Proceedings of the First USENIX Conference on Hot Topics in Parallelism (HotPar’09).
  * Gao et al. (2013) Wanling Gao, Yuqing Zhu, Zhen Jia, Chunjie Luo, and Lei Wang. 2013.  Bigdatabench: a big data benchmark suite from web search engines. 1–7.
  * Grandl et al. (2014) Robert Grandl, Ganesh Ananthanarayanan, Srikanth Kandula, Sriram Rao, and Aditya Akella. 2014\.  Multi-resource Packing for Cluster Schedulers, In Proceedings of the 2014 ACM Conference on SIGCOMM.  SIGCOMM Comput. Commun. Rev. 44, 4, 455–466.
  * Grewe et al. (2013) Dominik Grewe, Zheng Wang, and Michael FP O’Boyle. 2013\.  OpenCL task partitioning in the presence of GPU contention. In International Workshop on Languages and Compilers for Parallel Computing. 87–101.
  * Hauswald et al. (2015a) Johann Hauswald, Yiping Kang, Michael A. Laurenzano, Quan Chen, Cheng Li, Trevor Mudge, Ronald G. Dreslinski, Jason Mars, and Lingjia Tang. 2015a.  DjiNN and Tonic: DNN As a Service and Its Implications for Future Warehouse Scale Computers, In ISCA ’15.  SIGARCH Comput. Archit. News 43, 3, 27–40.
  * Hauswald et al. (2015b) Johann Hauswald, Michael A. Laurenzano, Yunqi Zhang, Cheng Li, Austin Rovinski, Arjun Khurana, Ronald G. Dreslinski, Trevor Mudge, Vinicius Petrucci, Lingjia Tang, and Jason Mars. 2015b.  Sirius: An Open End-to-End Voice and Vision Personal Assistant and Its Implications for Future Warehouse Scale Computers, In ASPLOS ’15.  SIGPLAN Not. 50, 4, 223–238.
  * Hindman et al. (2011) Benjamin Hindman, Andy Konwinski, Matei Zaharia, Ali Ghodsi, Anthony D. Joseph, Randy Katz, Scott Shenker, and Ion Stoica. 2011\.  Mesos: A Platform for Fine-grained Resource Sharing in the Data Center. In Proceedings of the 8th USENIX Conference on Networked Systems Design and Implementation (NSDI’11). 295–308.
  * Huang et al. (2011) Shengsheng Huang, Jie Huang, Jinquan Dai, Tao Xie, and Bo Huang. 2011.  The HiBench Benchmark Suite: Characterization of the MapReduce-Based Data Analysis. In New Frontiers in Information and Software as Services: Service and Application Design Challenges in the Cloud, Divyakant Agrawal, K. Selçuk Candan, and Wen-Syan Li (Eds.). Springer Berlin Heidelberg, Berlin, Heidelberg, 209–228.
  * Jacobs et al. (1991) Robert A. Jacobs, Michael I. Jordan, Steven J. Nowlan, and Geoffrey E. Hinton. 1991.  Adaptive Mixtures of Local Experts.  Neural Comput. 3, 1 (March 1991), 79–87.
  * Jiang et al. (2014) Tao Jiang, Qianlong Zhang, Rui Hou, Lin Chai, Sally A Mckee, Zhen Jia, and Ninghui Sun. 2014.  Understanding the behavior of in-memory computing workloads. In Workload Characterization (IISWC), 2014 IEEE International Symposium on Workload Characterization. IEEE, 22–30.
  * Keller and Gray (1985) James M. Keller and Michael R. Gray. 1985.  A Fuzzy K-Nearest Neighbor Algorithm.  IEEE Transactions on Systems, Man and Cybernetics (1985).
  * Kyrola et al. (2012) Aapo Kyrola, Guy Blelloch, and Carlos Guestrin. 2012\.  GraphChi: Large-scale Graph Computation on Just a PC. In Proceedings of the 10th USENIX Conference on Operating Systems Design and Implementation (OSDI’12).
  * Li et al. (2015) Min Li, Jian Tan, Yandong Wang, Li Zhang, and Valentina Salapura. 2015.  SparkBench: A Comprehensive Benchmarking Suite for in Memory Data Analytic Platform Spark. In Proceedings of the 12th ACM International Conference on Computing Frontiers (CF ’15).
  * Liu and Li (2014) Ming Liu and Tao Li. 2014\.  Optimizing Virtual Machine Consolidation Performance on NUMA Server Architecture for Cloud Workloads.  SIGARCH Comput. Archit. News 42, 3, 325–336.
  * Llull et al. (2017) Qiuyun Llull, Songchun Fan, Seyed Majid Zahedi, and Benjamin C Lee. 2017.  Cooper: Task Colocation with Cooperative Games. In High Performance Computer Architecture (HPCA), 2017 IEEE International Symposium on. 421–432.
  * Lo et al. (2015) David Lo, Liqun Cheng, Rama Govindaraju, Parthasarathy Ranganathan, and Christos Kozyrakis. 2015\.  Heracles: Improving Resource Efficiency at Scale. In Proceedings of the 42Nd Annual International Symposium on Computer Architecture (ISCA ’15).
  * Luk et al. (2009) Chi-Keung Luk, Sunpyo Hong, and Hyesoon Kim. 2009.  Qilin: Exploiting Parallelism on Heterogeneous Multiprocessors with Adaptive Mapping. In Proceedings of the 42Nd Annual IEEE/ACM International Symposium on Microarchitecture (MICRO 42). ACM, 45–55.
  * Manly (2004) Bryan FJ Manly. 2004\.  Multivariate statistical methods: a primer.  CRC Press.
  * Mashayekhy et al. (2015) L. Mashayekhy, M. M. Nejad, D. Grosu, Q. Zhang, and W. Shi. 2015.  Energy-Aware Scheduling of MapReduce Jobs for Big Data Applications.  IEEE Transactions on Parallel and Distributed Systems 26, 10 (2015), 2720–2733.
  * Nguyen et al. (2015) Khanh Nguyen, Kai Wang, Yingyi Bu, Lu Fang, Jianfei Hu, and Guoqing Xu. 2015\.  FACADE: A Compiler and Runtime for (Almost) Object-Bounded Big Data Applications. In Proceedings of the Twentieth International Conference on Architectural Support for Programming Languages and Operating Systems (ASPLOS ’15).
  * Ogilvie et al. (2017) William F Ogilvie, Pavlos Petoumenos, Zheng Wang, and Hugh Leather. 2017.  Minimizing the cost of iterative compilation with active learning. In Code Generation and Optimization (CGO), 2017 IEEE/ACM International Symposium on. 245–256.
  * Ousterhout et al. (2015) Kay Ousterhout, Ryan Rasti, Sylvia Ratnasamy, Scott Shenker, and Byung-Gon Chun. 2015.  Making Sense of Performance in Data Analytics Frameworks. In Proceedings of the 12th USENIX Conference on Networked Systems Design and Implementation (NSDI’15). USENIX Association, Berkeley, CA, USA, 293–307.
  * Ren et al. (2017) Jie Ren, Ling Gao, Hai Wang, and Zheng Wang. 2017\.  Optimise web browsing on heterogeneous mobile platforms: a machine learning based approach. In IEEE International Conference on Computer Communications (INFOCOM), 2017 (INFOCOM 2017).
  * Rupprecht et al. (2017) Lukas Rupprecht, William Culhane, and Peter R. Pietzuch. 2017\.  SquirrelJoin: Network-Aware Distributed Join Processing with Lazy Partitioning.  PVLDB 10, 11 (2017), 1250–1261.
  * Shvachko et al. (2010) Konstantin Shvachko, Hairong Kuang, Sanjay Radia, and Robert Chansler. 2010.  The Hadoop Distributed File System. In Proceedings of the 2010 IEEE 26th Symposium on Mass Storage Systems and Technologies (MSST) (MSST ’10). 1–10.
  * Simon et al. (2013) Douglas Simon, John Cavazos, Christian Wimmer, and Sameer Kulkarni. 2013.  Automatic Construction of Inlining Heuristics Using Machine Learning. In Proceedings of the 2013 IEEE/ACM International Symposium on Code Generation and Optimization (CGO) (CGO ’13). 1–12.
  * Singh et al. (2009) Karan Singh, Major Bhadauria, and Sally A. McKee. 2009\.  Real Time Power Estimation and Thread Scheduling via Performance Counters.  SIGARCH Comput. Archit. News 37, 2 (jul 2009), 46–55.
  * Sparks et al. (2013) Evan R Sparks, Ameet Talwalkar, Virginia Smith, Jey Kottalam, Xinghao Pan, Joseph Gonzalez, Michael J Franklin, Michael I Jordan, and Tim Kraska. 2013.  MLI: An API for distributed machine learning. In Data Mining (ICDM), 2013 IEEE 13th International Conference on. 1187–1192.
  * Stock et al. (2012) Kevin Stock, Louis-Noël Pouchet, and P. Sadayappan. 2012.  Using Machine Learning to Improve Automatic Vectorization.  ACM Trans. Archit. Code Optim. 8, 4, Article 50 (jan 2012), 23 pages.
  * Tang et al. (2011) Lingjia Tang, Jason Mars, Neil Vachharajani, Robert Hundt, and Mary Lou Soffa. 2011.  The Impact of Memory Subsystem Resource Sharing on Datacenter Applications.  SIGARCH Comput. Archit. News 39, 3, 283–294.
  * Taylor et al. (2017) Ben Taylor, Vicent Sanz Marco, and Zheng Wang. 2017.  Adaptive Optimization for OpenCL Programs on Embedded Heterogeneous Systems. In Proceedings of the 18th ACM SIGPLAN/SIGBED Conference on Languages, Compilers, and Tools for Embedded Systems (LCTES 2017). 11–20.
  * Thusoo et al. (2009) Ashish Thusoo, Joydeep Sen Sarma, Namit Jain, Zheng Shao, Prasad Chakka, Suresh Anthony, Hao Liu, Pete Wyckoff, and Raghotham Murthy. 2009.  Hive: A Warehousing Solution over a Map-reduce Framework.  Proc. VLDB Endow. 2, 2, 1626–1629.
  * Tournavitis et al. (2009) Georgios Tournavitis, Zheng Wang, Björn Franke, and Michael F.P. O’Boyle. 2009.  Towards a Holistic Approach to Auto-parallelization: Integrating Profile-driven Parallelism Detection and Machine-learning Based Mapping. In Proceedings of the 30th ACM SIGPLAN Conference on Programming Language Design and Implementation (PLDI ’09). 177–187.
  * Vavilapalli et al. (2013) Vinod Kumar Vavilapalli, Arun C. Murthy, Chris Douglas, Sharad Agarwal, Mahadev Konar, Robert Evans, Thomas Graves, Jason Lowe, Hitesh Shah, Siddharth Seth, Bikas Saha, Carlo Curino, Owen O’Malley, Sanjay Radia, Benjamin Reed, and Eric Baldeschwieler. 2013.  Apache Hadoop YARN: Yet Another Resource Negotiator. In Proceedings of the 4th Annual Symposium on Cloud Computing (SOCC ’13).
  * Verma et al. (2012) A. Verma, L. Cherkasova, and R. H. Campbell. 2012.  Two Sides of a Coin: Optimizing the Schedule of MapReduce Jobs to Minimize Their Makespan and Improve Cluster Performance. In 2012 IEEE 20th International Symposium on Modeling, Analysis and Simulation of Computer and Telecommunication Systems. 11–18.
  * Wang et al. (2015) Zheng Wang, Dominik Grewe, and Michael FP O’boyle. 2015\.  Automatic and portable mapping of data parallel programs to opencl for gpu-based heterogeneous systems.  ACM Transactions on Architecture and Code Optimization (TACO) 11, 4 (2015), 42.
  * Wang and O’Boyle (2009) Zheng Wang and Michael O’Boyle. 2009.  Mapping Parallelism to Multi-cores: A Machine Learning Based Approach. In Proceedings of the 14th ACM SIGPLAN Symposium on Principles and Practice of Parallel Programming (PPoPP ’09). 75–84.
  * Wang and O’Boyle (2010) Z. Wang and M. F. P. O’Boyle. 2010.  Partitioning streaming parallelism for multi-cores: A machine learning based approach. In 19th International Conference on Parallel Architectures and Compilation Techniques (PACT) (PACT ’10). 307–318.
  * Wang and O’boyle (2013) Zheng Wang and Michael F. P. O’boyle. 2013.  Using Machine Learning to Partition Streaming Programs.  ACM Trans. Archit. Code Optim. 10, 3 (2013), 20:1–20:25.
  * Wang et al. (2014) Zheng Wang, Georgios Tournavitis, Björn Franke, and Michael F. P. O’boyle. 2014.  Integrating Profile-driven Parallelism Detection and Machine-learning-based Mapping.  ACM Trans. Archit. Code Optim. 11, 1 (2014), 2:1–2:26.
  * Wen et al. (2014) Yuan Wen, Zheng Wang, and Michael FP O’Boyle. 2014.  Smart multi-task scheduling for OpenCL programs on CPU/GPU heterogeneous platforms. In High Performance Computing (HiPC), 2014 21st International Conference on. IEEE, 1–10.
  * Xu et al. (2015) Cong Xu, Brendan Saltaformaggio, Sahan Gamage, Ramana Rao Kompella, and Dongyan Xu. 2015\.  vRead: Efficient Data Access for Hadoop in Virtualized Clouds. In Proceedings of the 16th Annual Middleware Conference (Middleware ’15). ACM, New York, NY, USA, 125–136.
  * Xu et al. (2016) L. Xu, M. Li, L. Zhang, A. R. Butt, Y. Wang, and Z. Z. Hu. 2016.  MEMTUNE: Dynamic Memory Management for In-Memory Data Analytic Platforms. In 2016 IEEE International Parallel and Distributed Processing Symposium (IPDPS). IEEE, 383–392.
  * Yang et al. (2013) Hailong Yang, Alex Breslow, Jason Mars, and Lingjia Tang. 2013.  Bubble-flux: Precise Online QoS Management for Increased Utilization in Warehouse Scale Computers. In Proceedings of the 40th Annual International Symposium on Computer Architecture (ISCA ’13).
  * Yang et al. (2015) Xi Yang, Stephen M. Blackburn, and Kathryn S. McKinley. 2015\.  Computer Performance Microscopy with Shim.  SIGARCH Comput. Archit. News 43, 3, 170–184.
  * Zaharia et al. (2010) Matei Zaharia, Mosharaf Chowdhury, Michael J. Franklin, Scott Shenker, and Ion Stoica. 2010\.  Spark: Cluster Computing with Working Sets. In Proceedings of the 2Nd USENIX Conference on Hot Topics in Cloud Computing (HotCloud’10). 10–10.
  * Zhuravlev et al. (2010) Sergey Zhuravlev, Sergey Blagodurov, and Alexandra Fedorova. 2010\.  Addressing Shared Resource Contention in Multicore Processors via Scheduling.  SIGPLAN Not. 45, 3, 129–142.