Comparisons of Algorithms in Big Data Processing

Parallel computing for big data has different applications from online social networks, health-care industry, advertisement placement, and machine learning to search engines and data centers. The most popular big data parallel computing framework is MapReduce which is broadly used in Hadoop [White, 2012], Dryad [Isard et al., 2007], Google [Dean and Ghemawat, 2008], Deep Learning [Daghighi, 2019], and grid-computing [Isard et al., 2009, Saadatmand et al., 2019a, b]. Before talking about MapReduce, we present some details on the network structure of data centers. Data centers used to mainly consist of two parts, storage and computing, and these two parts were connected to each other through a network link with high bandwidth. When this structure was being used for normal data processing, the communication of data from storage to processing unit was not creating any bottleneck. However, with the emergence of Big Data, the network between the two units was unable to accommodate fast and reliable data transfer. Hence, scientists come up with the idea of bringing data into processing unit by splitting both data and processing units into hundreds of small units and combining each computing unit with a small storage unit. As a result, each small unit consisting of storage and processing units, called a server, is capable of storing small pieces of data and processing them at the same time. In other words, data does not need to be transferred from storage unit to processing unit since they are already together. However, note that the big data cannot be stored in the storage unit of a single server. The solution is to split the big data into small chunks of data, typically 68-128 MB, and storing them on multiple servers. In practice, each data chunk is stored on three servers to increase availability and decrease data loss probability. As a result, processing of big data consists of processing of multiple data chunks that make the whole big data, and concatenating the data chunk processing results for the completion of the big data process. The processing of each data chunk is called Map task and the concatenation of the results on all the Map tasks is called Reduce task, which make up the MapReduce processing framework for big data. Note that at least for Reduce tasks, servers need to be connected and cannot completely be isolated. We later see that even for executing Map tasks, servers may need to exchange data. Hence, servers are connected to each other through links and switches. The structure of switches connecting servers is a complete field of research in computer science, but servers are generally connected to each other by top of the rack switches as well as core switches in the following way. The hundred servers are grouped into batches of 20-50 servers, where each batch is inter-connected to each other by a switch, called rack switch, and all rack switches are connected to one or more core switches which make all the servers connected.

The rack structure of data centers brings a lot of complexity for load balancing. As a result, most theoretical work on load balancing for data centers either consider homogeneous model of servers or ignore the rack structure and only consider data locality, where data locality refers to the fact that the data chunk associated to a Map task is stored on three servers, so is not available on other servers immediately. Examples of works that consider homogeneous server model are [Livny and Melman, 1982, Singh, 1970, Kreimer, 2002], and [Lowery et al., 2008]. A branch of research on homogeneous servers is utilization of the power of two or more choices for load balancing that lowers the messaging overhead between hundreds of servers and the core load balancing scheduler. For example you can refer to [Dahlgaard et al., 2016, Mitzenmacher, 2001, Gardner et al., 2017, Richa et al., 2001, Byers et al., 2004, Doerr et al., 2011, Lumetta and Mitzenmacher, 2007, Cooper et al., 2014, Luczak et al., 2005], and [Gast, 2015]. Although there has been a huge body of work on heuristic algorithms for heterogeneous server model, examples of which are [White, 2010, Isard et al., 2009, Zaharia et al., 2010, Jin et al., 2011, He et al., 2011, Polo et al., 2011, Zaharia et al., 2008, Daghighi and Kavousi, 2017, Kavousi, 2017, Moaddeli et al., 2019], and [Salehi, 2017], there has been a few recent works on algorithms with theoretical guarantees on such more complicated models.

The scheduling problem for a data center with a rack structure is a specific case of the open affinity scheduling problem, where each task type can be processed by each server but with different processing rates. The classic MaxWeight algorithm [van de Ven et al., 2009, Sadiq and De Veciana, 2009, Stolyar et al., 2004, Meyn, 2009], c-$\mu$-rule [Van Mieghem, 1995, Mandelbaum and Stolyar, 2004], and the work by Harrison and López [1999], Harrison [1998], and by Bell et al. [2001, 2005] have different approaches on solving the load balancing problem for the affinity scheduling, but they either not solve the delay optimality or have unrealistic assumptions including known task arrival rates and existence of one queue per task type. The state-of-the-art on scheduling for data centers considering the rack structure, no knowledge of task arrival rates, and having queues on the order of servers not the number of task types is presented by Xie et al. [2016] and by Yekkehkhany [2017], which is extended for a general number of data locality levels by Yekkehkhany et al. [2018]. The central idea to all algorithms in [Xie et al., 2016, Yekkehkhany, 2017], and [Yekkehkhany et al., 2018] is to use weighted workload on servers instead of the queue lengths, which leads to a better perfromance in terms of average delay expereinced by submitted tasks. The Balanced-PANDAS alborithm, where PANDAS stands for Priority Algorithm for Near-Data Scheduling, is the name for the weighetd-workload based algorithms proposed in [Xie et al., 2016, Yekkehkhany, 2017], and [Yekkehkhany et al., 2018]. The Join-the-Shortest-Queue-MaxWeight (JSQ-MaxWeight, JSQ-MW) proposed by Wang et al. [2013] that only considers data locality is also extended to the case where rack structure is considered by Xie et al. [2016]. The priority algorithm proposed by Xie and Lu [2015] is another work that only considers data locality, not the rack structure of data centers which is interesting in its own rights since both throughput and heavy-traffic delay optimality are proved for it; however, it is not even throughput optimal for a system with rack structure.

All of the algorithms mentioned above consider complete knowledge about the processing rates of different task types on different servers. However, the reality is that the processing rates are mostly not known due to errors in estimation methods and the change of the system structure over time or the change of traffic which can change the processing rates. Hence, it is important that the algorithm that is used for load balancing is robust to estimation errors of processing rates. In this work, we run extensive simulations to evaluate the robustness of the state-of-the-art algorithms on load balancing with different levels of data locality. It is observed that the Balanced-PANDAS algorithm not only has a better heavy-traffic delay performance, but it also is more robust to changes of processing rate estimations, while MaxWeight based algorithm does not perform as well as Balanced-PANDAS under processing rate estimation errors. In order to estimate the processing rates of tasks on servers and better model the data center structure, reinforcement learning methods can be used as it is discussed in [Musavi et al., 2016b, Yildiz et al., 2013, Musavi et al., 2016a], and [Musavi, 2019]. A recent work by Yekkehkhany and Nagi [2020] considers an exploration-exploitation approach as in the reinforcement learning method to both learn the processing rates and exploit load balancing based on the current estimation of the processing rates. They propose the Blind GB-PANDAS algorithm that is proven to be throughput-optimal and have a lower mean task completion time than the existing methods. A more sophisticated risk-averse exploration-exploitation approach can be considered for this problem when different tasks have different risk-levels as discussed in [Yekkehkhany et al., 2019] and [Yekkehkhany et al., 2020].

The rest of the paper is organized as follows. Section 2 presents the system model that is used throughout the paper, section 3 summarizes the preliminary materials including the description of Balanced-PANDAS, Priority, and MaxWeight based algorithms that are needed before we present the robustness comparison among different algorithms in section 4.

We consider the same system model described in [Xie et al., 2016] and [Yekkehkhany et al., 2018] as follows. A discrete time model is considered, where time is indexed by $t\in\mathbb{N}$. Assume a data center with $M$ servers and denote the set of all servers as $\mathcal{M}=\{1,2,\cdots,M\}$. Without loss of generality, assume that the first $M_{R}$ servers are connected to each other with a top of the rack switch and are called the first rack, the second $M_{R}$ servers are connected to each other with another top of the rack switch and are called the second rack, and so on. Hence, there are $N_{R}=\frac{M}{M_{R}}$ racks in total. All the top of the rack switches are connected to each other with one or more core switches in a symmetric manner. As a result, there are three levels of data locality as described below. Recall that the data chunk associated to a map task is stored on three servers by Hadoop’s default, so all those three servers are called local servers for the map task or in other words the map task can receive service locally from those three servers. Since servers have the data chunk of local tasks, the processing is immediately started after servers are assigned to process them. Note that the three servers storing the data for a map task is normally different for different map tasks. Hence, we associate a type to each map task, which is the label of the three servers, i.e. $(m_{1},m_{2},m_{3})\in\mathcal{M}^{3},$ such that $m_{1}<m_{2}<m_{3}$. This gives us a unique and informative way of representation for different task types as follows:

where a task type is denoted by $\bar{L}=(m_{1},m_{2},m_{3})$ given that $m_{1},m_{2},$ and $m_{3}$ are the three local servers for task of type $\bar{L}$ and the set of all task types is denoted by $\mathcal{L}$.

A map task is not limited to receive service from one of the local servers. It can receive service from one of the servers that are in the same rack as the local servers with a slightly lower service rate. The slight depreciation of processing rate for such servers, which are called rack-local servers, is for the travel time of the data associated to a map task from a local server to the rack-local server that is assigned for processing the map task rack-locally. Finally, all other servers other than the local and rack-local servers, which are called remote servers, have the lowest processing rate for a map task, since data needs to be transmitted through at least two of the top rack switches and a core switch, so the server cannot immediately start processing the task when it is assigned to do so remotely. In order to formally define the rack-local and remote servers, we need to propose a notation. Let $R(m)\in\{1,2,\cdots,N_{R}\}$ denotes the label of the rack that the $m$-th server belongs to. Then, the set of rack-local and remote servers to task of type $\bar{L}=(m_{1},m_{2},m_{3})$, denoted by $\bar{L}_{k}$ and $\bar{L}_{r}$, respectively, are as follows:

The service and arrival process of tasks is described below.
Task arrival process: Let $A_{\bar{L}}(t)$ denote the number of tasks of type $\bar{L}$ that arrive to the system at time slot $t$, where $\mathbb{E}[A_{\bar{L}}(t)]=\lambda_{\bar{L}},$ and it is assumed that $A_{\bar{L}}(t)<C_{A}$ and $P(A_{\bar{L}}(t)=0)>0$. The set of arrival rates for all task types is denoted by $\mathbf{\lambda}=(\lambda_{\bar{L}}:\bar{L}\in\mathcal{L})$.
Service process: The processing of a task on a local server is assumed to be faster than on a rack-local server, and the processing of a task on a rack-local server is faster than on a remote server. This fact is formalized as follows. The processing time of a task on a local, rack-local, and remote server has means $\frac{1}{\alpha},\frac{1}{\beta},$ and $\frac{1}{\gamma},$ respectively, where $\alpha>\beta>\gamma.$ Note that the processing time of a task can have any distribution with the given means, but the heavy-traffic delay optimality of Balanced-PANDAS algorithm is only proven under Geometric service time distribution, while MaxWeight based algorithm does not have a general heavy-traffic delay optimality under any distribution for service time.
Capacity region characterization: An arrival rate for all task types is supportable for service by the $M$ servers if and only if the load on each server is strictly less than the capacity of the server. Considering a processing rate of one for each server, an arrival rate vector $\mathbf{\lambda}=(\lambda_{\bar{L}}:\bar{L}\in\mathcal{L})$ is in the capacity region of the system if and only if:

where $\lambda_{\bar{L},m}$ is the rate of incoming tasks of type $\bar{L}$ that are processed by server $m$.

In this section, we introduce the main three algorithms on scheduling for data centers with more than or equal to two levels of data locality that have theoretical guarantees on optimality in some senses and under some conditions. In order to introduce the load balancing algorithm of each method, we also need to present the queueing structure required for that method. The three algorithms are

1. 1.

   Priority algorithm [Xie and Lu, 2015], which is best fit for applications with two levels of data locality, e.g. for the cases that only data locality is taken into account. An example is scheduling for Amazon Web Services inside a rack.

2. 2.

   Balanced-PANDAS [Xie et al., 2016, Yekkehkhany, 2017], and [Yekkehkhany et al., 2018], which is the state-of-the-art for scheduling applications with multiple levels of data locality and is observed to perform better in terms of average task completion time by fourfold in comparison to MaxWeight based algorithms. It is proven by [Xie et al., 2016] that under mild conditions, Balanced-PANDAS is both throughput and heavy-traffic delay optimal for a system with three levels of data locality and a rack structure.

3. 3.

   MaxWeight based algorithms [Stolyar et al., 2004] and [Wang et al., 2016], which can be used for multiple levels of data locality and are throughput optimal, but not heavy-traffic delay optimal, and it is observed that they generally have poor performance at high loads compared to weighted workload based algorithm used in Balanced-PANDAS algorithm.

The following three subsections present a complete introduction to these three main algorithms.

In this section, we present the results on our extensive simulations on robustness of scheduling algorithms presented in section 3. To this end, we run the algorithms with parameters that have error to study which algorithm can tolerate errors better than others. More specifically, we use incorrect $\alpha,\beta,$ and $\gamma$ in the algorithms for calculating weighted workloads or weighted queue lengths and observe the average task completion time under these scenarios. The arrival process is a Poisson process and the processing time has an exponential distribution. We have also tested the algorithms for processing times with heavy-tailed distributions and observed similar results. We make these parameters $5\%,10\%,15\%,20\%,25\%,$ and $30\%$ off their real value, either greater than the real value or smaller than the real value, and evaluate algorithms under these cases. The traffic load is assumed the same under all algorithms so that the comparison makes sense. We further compare all the three algorithms mentioned in section 3 with the Hadoop’s default scheduler which is First-In-First-Out (FIFO). Figure 1 shows the comparison between the four algorithms when precise value of parameters are known by the central scheduler. As we see, Balanced-PANDAS algorithm has the lowest average task completion time at high loads. A closer look of high loads is presented in figure 2, where Balanced-PANDAS obviously outperform JSQ-MaxWeight in terms of average task completion time.

The performance of the four algorithms are compared to each other when the parameters are lower than their real values by certain percentages, where the results are shown in figure 3. As is seen, Balanced-PANDAS has best performance among all algorithms by changing the parameters’ error from 5% to 30%. In fact, figure 4 shows that the Balanced-PANDAS has the least sensitivity against change of parameters while JSQ-MaxWeight’s performance varies notably by the increase of error in parameter estimations.

Comparison of the algorithms when the parameters are off for some percentages, but higher than their real values are given in figure 5. It is again observed that the Balanced-PANDAS algorithm has consistent better performance than the JSQ-MaxWeight algorithm. The sensitivity comparison of the Balanced-PANDAS and JSQ-MaxWeight algorithms in this case is presented in figure 6.

In this work, we did a literature review on both classical and state-of-the-art scheduling algorithms for the affinity scheduling problem. Data center load balancing is a special case of the affinity scheduling problem. Considering the rack structure of data centers, there are three levels of data locality. The priority algorithm that is heavy-traffic delay optimal is not even throughput optimal for three levels of data locality. The Balanced-PANDAS algorithm is the state-of-the-art in heavy-traffic delay optimality. We investigated the robustness of Balanced-PANDAS and JSQ-MaxWeight algorithms with respect to errors in parameter estimation. We observe that Balanced-PANDAS keeps its better performance even in the absence of precise parameter values versus JSQ-MaxWeight. Note that the JSQ-MaxWeight algorithm is also robust under parameter estimation errors, but it is more sensitive than Balanced-PANDAS, specially at high loads close to the boundary of the capacity region. For future work, one can use machine learning tools to estimate the system parameters and make them more precise in the meanwhile that the load balancing algorithm is working with the estimated parameters. The scheduling algorithms presented in this work can also be applied to a vast number of applications including but not limited to healthcare and super market models [Winkler, 1987, Clower and Leijonhufvud, 1975, Eisenhauer, 2001, Hosseini et al., 2017], web search engines [Schwartz, 1998, Salehi et al., 2018, Broder, 2002, Krishna and Rani, 2018, Xie et al., 2018], electric vehicle charging [Alinia et al., 2018, Gan et al., 2013, Wang et al., 2005, Deilami et al., 2011, Chehardeh and Hatziadoniu, 2018, Almalki et al., 2015, Chehardeh et al., 2016, Saadatmand et al., 2020a, b] and so on.