Cost-based Data Prefetching and Scheduling in Big Data Platforms over Tiered Storage Systems

A typical compute cluster consists of a set of nodes \(N = \lbrace N_1, \ldots, N_r\rbrace\) arranged in a rack network topology. The cluster offers a set of resources \(R = \lbrace R_1, \ldots, R_m\rbrace\) for executing tasks. A resource represents a logical bundle of physical resources (e.g., \(\langle\)1 CPU, 2 GB RAM\(\rangle\)), such as a Container in Apache YARN, an Executor slot in Spark, or a Resource offer in Apache Mesos, and it is bound to a particular node. Resources are dynamically created based on the availability of physical resources and the requirements of the tasks. For each resource \(R_j\), we define its location as \(L(R_j) = N_k\), where \(N_k \in N\). Finally, a set of tasks \(T = \lbrace T_1, \ldots, T_n\rbrace\) require resources for executing on the cluster.

In a traditional big data environment (i.e., in the absence of tiered storage), each task contains a list of preferred node locations based on the locality of the data it will process. For example, if a task will process a data block that is replicated on nodes \(N_1\), \(N_2\), and \(N_4\), then its list of preferred locations contains these three nodes. However, when the data are stored in a tiered storage system such as OctopusFS or tiered HDFS, the storage tier of each block is also available. Hence, we define the task’s preferred locations \(P(T_i) = [\langle N_k, p_k^i \rangle ]\) as a list of pairs, where the first entry in a pair represents the node \(N_k \in N\) and the second entry \(p_k^i \in \mathbb {R}\) represents the storage tier. We define \(p_k^i\) as a pre-defined numeric score that represents the cost of reading the data from that storage tier. Hence, the lower the score the better for performance. Various metrics can be used for setting the scores but their absolute values are not as important as representing the relative performance across the tiers. For example, if the I/O bandwidth of memory, SSD, and HDD media is 3,200, 400, and 160 MB/s, respectively, then the scores 1, 8, and 20 would capture the relative cost of reading data from those three tiers.

Scheduling a task \(T_i\) on a resource \(R_j\) will incur an assignment cost C based on the following cost function:

According to Equation (1), if the location of resource \(R_j\) is one of the nodes \(N_k\) in \(T_i\)’s preferred locations, then the cost will equal the corresponding tier preference score \(p_k^i\). Alternatively, if \(R_j\) is on the same rack as one of the nodes \(N_k\) in \(T_i\)’s preferred locations, then the cost will equal the corresponding preference score \(p_k^i\) plus a constant \(c_1\), which represents the network transfer cost within a rack. Otherwise, the cost will equal a constant \(c_2\), representing the network transfer cost across racks. The inter-rack network cost is often much higher than the intra-rack cost and dwarfs the local reading I/O cost; hence, we do not add a preference score to \(c_2\) in our current cost function. For modern clusters that use Remote Direct Memory Access (RDMA) over commodity Ethernet (RoCEv2) for intra data center communication [26], the preference score \(c_2\) can be set much closer to \(c_1\) and the cost function can be adjusted to also add the tier-based preference score \(p_k^i\). In general, the cost function can be easily adjusted to represent the task-to-resource assignment cost for a particular environment without affecting the task scheduling problem formulation or solution approach.

Using the above definitions, the task scheduling problem can be formulated as a constrained optimization problem:

The goal is to find all assignments (i.e., \(\langle T_i, R_j \rangle\) pairs) that will minimize the sum of the corresponding assignment costs. The variable \(x_{i,j}\) is 1 if \(T_i\) is assigned to \(R_j\) and 0 otherwise. The second constraint guarantees that each task will only be assigned to one resource, while the last constraint guarantees that each resource will only be assigned to one task. The above formulation requires that the number of tasks n equals the number of resources m. If \(n \lt m\), then the third constraint must be relaxed to \(\sum _{T_i \in T } x_{i,j} \le 1\) (i.e., some resources will not be assigned), while if \(n \gt m\), the second constraint must be relaxed to \(\sum _{R_j \in R } x_{i,j} \le 1\) (i.e., some tasks will not be assigned). Table 1 summarizes the notation used in this section.

The task scheduling problem defined above can be encoded as a bipartite graph \(G = (T, R, E)\). The vertex sets T and R correspond to the tasks and resources, respectively, and together form the vertices of graph G. Each edge \((T_i, R_j)\) in the edge set E connects a vertex \(T_i \in T\) to a vertex \(R_j \in R\) and has a weight (or cost) as defined in Equation (1). The constrained optimization problem formulated in Equation (2) is equivalent to finding a maximum matching \(M = \lbrace (T_i, R_j)\rbrace\) with \((T_i, R_j) \in T \times R\) in the bipartite graph G that minimizes the total cost function,

By definition, a matching is a subset of edges \(M \subseteq E\) such that for all vertices v in G, at most one edge of M is incident on v. Hence, at most one task will be assigned to one resource and vice versa. A maximum matching is a matching of maximum cardinality, i.e., it contains as many edges as possible. Since any task can potentially be assigned to any resource, maximum matching will contain either all tasks or all resources, depending on which set is smaller. Hence, the constraints listed in Equation (2) are all satisfied. Consequently, by solving the minimum cost maximum matching problem on G, we are guaranteed to assign as many tasks as possible to resources and to attain the lowest total assignment cost possible.

Figure 3 illustrates an example with three tasks (\(T_1\)–\(T_3\)) and four available resources (\(R_1\)–\(R_4\)) located on three distinct nodes (\(N_1\), \(N_2\), \(N_4\)). Each task \(T_i\) has a list of three preferred locations (i.e., \(\langle\)node, tier\(\rangle\) pairs) according to the storage location of the corresponding data block \(B_i\) shown in Figure 2. For ease of reference, each tier preference score is indicated using the constants M, S, and H, corresponding to the memory, SSD, and HDD tiers of the underlying storage system, with \(M \lt S \lt H\). Note that the definition of the preference score in Section 2.1 allows for variable values in case the cluster contains heterogeneous storage devices (e.g., SSDs with different I/O bandwidth on different nodes). The created bipartite graph with seven vertices (three for tasks and four for resources) is also visualized in Figure 3. Each edge corresponds to a potential assignment of a task to a resource and is annotated with the cost computed using Equation (1). The goal of task scheduling in this example is to select the three edges that form a maximum matching and minimize the total cost. The optimal task assignment (see highlighted edges in Figure 3) consists of two memory-local tasks (\(T_1\) on \(R_1\) and \(T_3\) on \(R_3\)) and one SSD-local task (\(T_2\) on \(R_4\)).

Several standard solvers can be used for solving the minimum cost maximum matching problem and finding the optimal task assignment, including the Simplex algorithm, the Ford–Fulkerson method, and the Hungarian Algorithm [18]. While the Simplex algorithm is known to perform well for small problems, it is not efficient for larger problems, because its pivoting operations become expensive. Its average runtime complexity is \(O(\mbox{max}(n, m)^3)\), where n is the number of tasks and m is the number of resources, but has a combinatorial worse case complexity. The Ford–Fulkerson method requires converting the problem into a minimum cost maximum flow problem, with complexity \(O((n+m) n m)\) in this setting. We have chosen to use the Hungarian Algorithm, as it runs in a strongly polynomial time with low hidden constant factors, which makes it more efficient in practice. Specifically, its complexity is \(O(nmx + x^2 \mbox{lg}(x))\), where \(x = \mbox{min}(n, m)\). Below, we introduce two vertex pruning algorithms that reduce the complexity to \(O(\mbox{min}(n, m)^3)\), while still guaranteeing an optimal solution. Approximation algorithms are also available for finding a near-optimal solution with a lower complexity [20]. However, the scheduling time of the overall approach is so low (as evaluated in Section 7.5) that we opted for finding the optimal solution with the Hungarian Algorithm.

In many cases, the number of tasks ready for execution does not equal the number of available resources. For example, a small job executing on a large cluster will have much fewer tasks than available resources, while a large job executing on a small or busy cluster will have much more tasks than available resources. Next, we describe two algorithms for pruning excess resources or tasks that reduce the graph size and lead to a more efficient execution of the Hungarian Algorithm, without affecting the optimality of the solution.

Algorithm 1 shows the process of computing and pruning the available resources in a cluster, which will then be used for building the bipartite graph and making the task assignments. The input consists of a list of tasks and a list of resource sets. A resource set represents a bundle of resources available on a node, which can be divided into resources (i.e., containers or slots) for running the tasks. The pruning of excess resources is enabled when the total available slots across all resource sets is d times higher than the number of tasks (line 3), where d is the default replication factor of the file system. The rationale for this limit will be explained after the algorithm’s description. First, the lists with the preferred locations of all tasks are traversed for counting the number of local tasks that can potentially be executed on each node (lines 4–7). The counts are stored in a map for easy reference. Next, each resource set \(S_j\) is considered (line 8). If \(S_j\) can be used to run at least one data-local task (line 9), then we need to compute the maximum number of slots (\(\mathit {maxSlots}\)) that can be created from \(S_j\) for running data-local tasks. \(\mathit {maxSlots}\) will equal the minimum of (a) the total number of available slots from \(S_j\) and (b) the number of tasks that contain \(S_j\)’s node in their preferred locations list (lines 10–12). Finally, \(\mathit {maxSlots}\) entries (i.e., resource slots from \(S_j\)) are added in the list of available resources (line 13). After traversing all resource sets, if the list of available resources has more entries than tasks, then the list is returned and the process completes (lines 14 and 15). Otherwise, resource slots for all remaining available resources are added in the result list (lines 16–18). This final step (lines 16–18) is also performed when the number of total available slots is less than d times the number of tasks for returning all available resources. Algorithm 1 is very efficient with a linear complexity of \(O(n+m)\), where n is the number of tasks and m is the total number of available resources.

Suppose 3 tasks are ready for execution and their preferred locations are as shown in Figure 4. Further, the cluster consists of 6 nodes (\(N_1\)–\(N_6\)), each with enough resources to create 3 slots. Hence, there are a total of 18 available slots and the pruning will take place. First, the number of possible data-local tasks will be computed as \(\lbrace N_1:2, N_2:2, N_3:2, N_4:3\rbrace\). For the resource set of \(N_1\), even though there are 3 available slots, only 2 will be added in the result list as only 2 can host data-local tasks. The same is true for the resource sets of \(N_2\) and \(N_3\). For \(N_4\), all 3 available slots will be added in the result list. Finally, since \(N_5\) and \(N_6\) do not appear in the tasks’ preferred locations, no slots will be added in the result list. Overall, only 9 of the 18 possible resource slots will be considered for the downstream task assignments. In fact, even if the number of available resource slots were much higher, 9 is the largest number of slots this process will return for this example. In general, n tasks and m resources (with \(n \ll m\)) will lead to a graph with \(n+m\) vertices and nm edges without pruning, but only \((1+d)n\) vertices and \(dn^2\) edges with pruning, showcasing that a massive pruning of excess resources is possible for small jobs.

Next, consider a different example where the same three tasks (with the same preferred locations) are present but only two slots are available on each of the nodes \(N_1\)–\(N_4\). If pruning were to take place (lines 3–15), then all eight slots would be added in the result list, rendering the pruning process pointless. This behavior is expected, since each task will typically have three preferred locations (since each file block has three replicas by default) spread across several nodes. Hence, by enabling pruning only when the number of available slots is greater than d times the tasks, we avoid going through a pruning process that will have no to little benefits.

Even though a large number of available resources may be pruned, the optimality of the task assignments is still guaranteed, because (a) the excluded resources cannot lead to data-local assignments and (b) the retained resources that can lead to data-local assignments are more than the tasks. Hence, the excluded resources would not have appeared in the final task assignments.

Algorithm 2 shows the process of pruning excess tasks in the presence of few available resources. In particular, pruning is enabled when the number of tasks ready for execution is higher than d times the total number of available resources (line 2). The rationale is similar to before: Avoid going through the process of pruning tasks when pruning is not expected to significantly reduce (if any) the number of tasks. When pruning is enabled, the available resources are traversed for collecting the set of distinct nodes (\(\mathit {resourceNodes}\)) they are located on (lines 3–5). Next, each task is added in the result list only if at least one of its preferred locations is contained in \(\mathit {resourceNodes}\) (lines 7–9). If the selected tasks are more than the available resources, then the result list is returned and the process completes (lines 10 and 11). In the opposite case, or when pruning is not enabled, the list with all tasks is returned (line 12). Similarly to Algorithm 1, Algorithm 2 is also very efficient with a linear complexity of \(O(n+m)\).

Continuing with the example with the three tasks (recall Figure 3), suppose that only one resource slot is available on node \(N_2\) as shown in Figure 5. In this case, only tasks \(T_1\) and \(T_3\) will be included in the selected tasks list as they record \(N_2\) in their preferred locations; task \(T_2\) will be excluded. The optimality of the task assignments is safeguarded, because the excluded tasks (which cannot lead to data-local assignments) would have never been selected by the Hungarian Algorithm, since there are still more (data-local) tasks than available resources.

Figure 6 shows the overall application flow over YARN. When a MapReduce or Spark application submits a job for execution ①, an Application Master (AM) is launched ② for managing the job’s lifecycle, negotiating resources from the Resource Manager (RM), and working with the Node Managers (NMs) to execute and monitor the tasks. First, the AM creates a set of resource requests ③ based on the input data to process and the tasks to execute. A resource request contains the following [59]:

The AM uses heartbeats (every 1 second by default) to send the requests to the RM and to receive the allocated containers. The RM also receives periodic heartbeats from NMs containing their resource availability. Upon a node heartbeat, the RM uses a pluggable Resource Scheduler to allocate the node resources to applications. The order of applications and the amount of resources to allocate to each one depend on the type of scheduler (e.g., First In First Out (FIFO) vs. Fair). Data locality is taken into account only when it is time to allocate resources to a particular application at a specific request priority. At that point, the scheduler tries to allocate data-local containers; otherwise, rack-local containers; otherwise, remote ones. The current schedulers also support the option of doing the allocations asynchronously (e.g., every 100 ms) based on all available resources in the cluster but do so in the same manner as described above.

Upon a heartbeat from an AM, the RM returns the allocated containers ④. The AM will then assign tasks to the allocated containers while taking data locality into account following the same strategy as above: first data-local, then rack-local, then remote. Next, the tasks are sent to the NMs for execution ⑤. When the number of tasks to execute is greater than the number of available containers, multiple waves of parallel tasks will get executed. In this case, whenever some tasks finish execution, the AM is notified ⑥, and upon the next heartbeat with the RM ⑦, it will receive new resource allocations ⑧. The AM will then assign tasks to the allocated containers as before and send them for execution at the NMs ⑨. When all tasks finish execution ⑩, the AM will terminate \({\bigcirc \!\!\!\!\!\scriptsize{11}}\), and the application will be notified about the job completion \({\bigcirc \!\!\!\!\!\scriptsize{12}}\).

Given a list of tasks with preferred locations, the AM will generate the resource requests as follows. For each distinct node N (or rack R) that appears in the preferred locations, a request will be created for N (or R), where the number of containers will equal the number of times N (or R) is found in the preferred locations. Finally, a ‘\(*\)’ (i.e., anywhere) request will be created, with the number of containers equal to the number of tasks.

The current Resource Schedulers are oblivious to the underlying tiered storage, since YARN’s resource requests do not include any storage tier preferences. To address this issue, we extend the resource request model to include a new preference map in the resource request, which maps each tier preference score (recall Section 2.1) to the number of containers requested for that score (i.e, tier). Consider the example in Figure 7 containing three tasks with preferred locations. Tasks \(T_1\) and \(T_2\) list node \(N_1\) in their preferred locations, which leads to the creation of one resource request for two containers on \(N_1\). The preference map will contain the entries \(\lbrace M:1,S:0,H:1\rbrace\), because \(N_1\) is paired with the score M for \(T_1\) and with H for \(T_2\). Regarding node \(N_2\), the preferred locations for \(T_1\) and \(T_3\) contain the pairs \(\langle N_2, S \rangle\) and \(\langle N_2, M \rangle\), respectively. Hence, the resource request for \(N_2\) contains two containers, with preference map \(\lbrace M:1,S:1,H:0\rbrace\). This information can then be used by the scheduler for making better decisions. For example, instead of allocating two containers on \(N_1\) (which would lead to two data-local containers, one memory-local, and one HDD-local), it would be better to allocate one container on \(N_1\) and 1 on \(N_2\), which would lead to two memory-local containers.

When computing the preference map for a rack-local resource request, we only count the lowest score (i.e., highest tier) for each task that is paired with nodes belonging to that rack. The rationale is to match the default behavior of the underlying tiered storage systems that direct a rack-local read to the highest tier (with the lowest score). In the example in Figure 7, all nodes belong to the same rack (\(r_1\)) and the lowest scores for \(T_1\), \(T_2\), and \(T_3\) are M, S, and M, respectively. Hence, the preference map contains \(\lbrace M:2,S:1,H:0\rbrace\). The same process is performed for computing the preference map for the ‘\(*\)’ (i.e., anywhere) request.

It is important to note that the current resource request model forms a “lossy compression of the application preferences” [59], which makes the communication and storage of requests more efficient, while allowing applications to express their needs clearly [59]. However, the exact preferred locations of the tasks cannot be mapped from the resource requests back to the individual tasks. Hence, the Trident Resource Scheduler will follow a different scheduling approach rather than using the Hungarian Algorithm, described next.

The Resource Scheduler is responsible for allocating resources to the various running applications by effectively making three decisions: (1) for which application to allocate resources next, (2) how many resources to allocate to that application, and (3) which available resources (i.e., containers) to allocate. The first two decisions are subject to various notions or constraints of capacities, queues, fairness, and so on [51]. For example, the Capacity Scheduler will select the first application of the most under-served queue and assign to it resources subject to that queue’s limit. Fair Scheduler, however, will select the application that is farthest below its fair share and assign resources up to its share limit. Locality is only taken into consideration during the third decision, when a specific amount of available resources are allocated to a particular application. Hence, the Trident’s resource allocation approach, which focuses only on the third decision, can be incorporated into a variety of existing schedulers, including FIFO, Capacity, and Fair, for making assignments based on both node locality and storage tier preferences.

Algorithm 3 shows Trident’s container allocation process, running in YARN’s Resource Manager. The input is a list of resource requests submitted by an application (with a particular priority) and the number of maximum containers to allocate based on queue capacity, fairness, and so on. The high level idea is to first build a list of potential containers to allocate based on locality preferences and then allocate the containers with the lowest assignment cost. The total number of containers is computed as the minimum of the total number of requested containers and the maximum allowed containers (line 3). Next, for each node-local resource request \(S_i^n\) that references a particular cluster node \(N_k\) (lines 4 and 5), we compute the number of available containers on \(N_k\) based on \(N_k\)’s available resources and \(S_i^n\)’s requested resources per container (line 6) as done in Apache YARN. The number of containers on \(N_k\) (\(\mathit {numConts}\)) will equal the minimum between the number of available containers and the number of requested containers in \(S_i^n\) (line 7). Finally, \(\mathit {numConts}\) containers will be added in the list of potential containers (line 8). The assignment cost for each container is also computed based on the preference map in \(S_i^n\) in procedure \(\mathit {AddContainers}\), which will be described later.

If the number of potential containers so far is less than the number of needed containers (line 9), then a similar process is followed for the resource requests with rack locality. Specifically, for each rack-local resource request (line 10) and for each node in the corresponding rack (line 11), the appropriate number of containers is added in the list of potential containers (lines 12–14). In this case, the network cost \(c_1\) is added to the assignment cost of each container (recall Equation (1)). As soon as the number of needed containers is reached, the double for loop is exited for efficiency purposes (lines 15 and 16). If the number of collected containers is still below the needed ones, then the remaining potential containers from random nodes are added in the list with assignment cost equal to \(c_2\) (lines 17 and 18).

Due to the aggregate form of the resource requests (recall Section 3.2), it is possible that the number of potential containers is higher than the needed containers (\(\mathit {totalConts}\)), even after the first loop iteration of node-local requests (lines 4–8). In common scenarios where data blocks are replicated 3 times, this number will typically equal 3 times \(\mathit {totalConts}\). This behavior is desirable for considering all storage tier preferences of the requests. Hence, the last step is to select the \(\mathit {totalConts}\) containers with the smallest assignment cost from the list of potential containers (lines 19 and 20). The sort on line 19 dominates the complexity of Algorithm 3 as \(O(k lg(k))\), where k is the number of potential containers, which typically equals 3 times the number of requested containers.

The \(\mathit {AddContainers}\) procedure (also shown in Algorithm 3) is responsible for creating and adding a number of potential containers on a node. The assignment costs for each container depend on the tier preferences of the particular resource request. The key intuition of the algorithm (lines 22–30) is to assign the lowest cost first as many times as it is requested. Next, assign the second lowest cost as many times as requested, and so on. For example, suppose the preference map contains \(\lbrace M:2,S:3,H:1\rbrace\) and three containers are needed. The three corresponding assignment costs to containers will equal M, M, and S.

Figure 7 shows a complete example with the resource requests generated based on the tasks preferred locations and the current available resources in a cluster with six nodes. The resource request on \(N_1\) contains two containers, which can fit in the available resources of \(N_1\). Hence, two potential containers are created, one with cost M and one with cost H (per the request’s preference map). The request on \(N_2\) also asks for two containers but only one can fit there; thus, one container is created with cost M. There are no available resources on \(N_3\) so no containers are created there. Finally, even though four containers can fit on \(N_4\), the corresponding request asks for only three, which leads to the creation of three potential containers with costs S, S, and H. At this point, the list of potential containers contains six entries, which are more than the three total requested containers. Finally, this list is sorted based on increasing assignment cost and the first three containers are allocated to the application, leading to the best resource allocation based on preferred (node and storage tier) locality.

The models presented in this section can be used to determine the performance of a job when a specific set of tasks is planned for execution over a specific set of available cluster resources and a specific set of input data is planned to be prefetched. For this purpose, two models are developed: one that simulates the prefetching of input data on the underlying file system and one that simulates the parallel execution of tasks on available cluster resources. The models’ novelty, efficiency, and accuracy come from how they use a mix of simulation and cost-based estimation at the level of individual input data blocks and tasks.

Algorithm 4 presents the prefetching model. The input to the model is a list of blocks to prefetch into memory from specific source storage devices. This model will set the time when each block will become available in the cache. Given that prefetching happens using concurrent I/O threads that start about the same time, prefetching more than one block per source device will lead to disk contention. Hence, unlike past modeling works (e.g., References [52, 64]), our model takes disk contention into consideration. The first step is to group the blocks based on the source device (line 2). Then, for each device, the blocks to be prefetched are retrieved and sorted on increasing block length (lines 3 and 4). Suppose that n blocks (\(\mathit {numBlocks}\) on Algorithm 4) will be prefetched from a specific device (line 5). At first, the n blocks will share the I/O bandwidth of the device equally. The block with the smallest length, \(b_1\), is expected to complete prefetching first. The time needed for \(b_1\)’s prefetching will equal its length divided by its share of the bandwidth, i.e., total bandwidth BW divided by n (line 10). During this time, the other blocks have made equal progress toward completion. Thus, the block with the second smallest length, \(b_2\), is expected to complete next, but its remaining length to be prefetched (line 9) will only compete with \((n-1)\) I/O threads (line 14). Hence, the total time needed for \(b_2\)’s prefetching will equal the time to prefetch \(b_1\) plus the time to prefetch its remaining length (with the bandwidth divided by \(n-1\)). The algorithm continues in a similar fashion for all remaining blocks (lines 8–14) until it sets the cache times for all prefetched blocks. Note that in practice, the device bandwidth mightnot be shared equally all the time across concurrent readers due to operating system scheduling. However, the data blocks read are typically very large (up to 128 MB), smoothing out any variations in I/O times across the readers. While linearly dividing the bandwidth among concurrent I/O threads is a reasonable simplifying assumption in our work, this part can be easily extended with a more expressive storage model that captures modern devices, such as the Parametric I/O Model [47]. Finally, the model also accounts for a constant initialization time (\(\mathit {TimeToInitCache}\)) that represents the time needed for scheduling prefetching requests and depends on the heartbeat intervals used between the file system components (defaults to 1 second).

The bottom half of Figure 8 shows two examples of modeling prefetching. In Figure 8(a), only one block is prefetched per storage device. For instance, suppose block \(B_3\) has length 128 MB and the device’s bandwidth is 32 MB/s. Hence, it will take 4 seconds for caching to complete. In Figure 8(b), two blocks are prefetched from the second storage device; \(B_2\) with length 96 MB and \(B_3\) with length 128 MB. Thus, the effective bandwidth for prefetching \(B_2\) is 16 MB/s, and \(B_2\) will take 6 seconds to be prefetched. At the same time, the first 96 MB of \(B_3\) will also be prefetched. After that, the remaining 32 MB of \(B_3\) will utilize the full bandwidth and prefetching will complete after an additional 1 second.

Algorithm 5 shows the modeling of parallel task execution on the available cluster resources. The input consists of a set of tasks submitted for execution (each to process an input data block), information about the available cluster resources, and a delay time parameter that will be explained at the end of the section. This method will set the expected start time and end time of each task based on the available resources and the storage type to read from. The number of tasks for execution is often larger than the number of available resource slots, causing the tasks to be executed in multiple waves. To account for this fact, a min priority queue of slots is created to keep track of when each slot is becoming available for hosting the next task for execution (line 2). In addition, the tasks are also placed in a min priority queue and the priority key is set to the tier preference score (recall Section 2.1) to simulate the presence of a tier-storage-aware task scheduler. It is important to note that a task planned to process a data block that is scheduled for prefetching (but prefetching has not completed yet) will get a tier preference score higher than the slowest tier but lower than the rack-local cost (for example, a score between the HDD score H and the network transfer cost \(c_1\)). Once prefetching completes, the preference score will be adjusted to reflect the memory tier. In this manner, a task scheduler will prioritize scheduling other data-local tasks first and leave the tasks with prefetched data to get scheduled later, after prefetching has completed.

For each task (taken in order of the tier preference score) and for each available slot (lines 4 and 5), Algorithm 5 will first compute the start time of the task (lines 6–9). The start time depends on the end time of the previous task executed on the current slot (if any) plus some constant times that represent the job initialization time (\(\mathit {TimeToInit}\)) and scheduling time (\(\mathit {TimeToSchedule}\)). Both constants are set based on the heartbeat intervals of the platform’s components and are set by default to 2 and 1 seconds, respectively. Next, the algorithm computes the task’s expected I/O duration (lines 10–13). If the task will start execution after the caching of its block is completed, then the task will read the block from memory; otherwise, it will read the block from its base storage device. For calculating the CPU processing time of a task, a simple linear regression model is utilized as a function of the input data size, which is built based on a sample of previous task execution logs (line 14). More advanced models for estimating the task processing time, such as the ones proposed in References [14, 31, 54], can be easily plugged in. The current slot is then updated with the tasks’ end time and placed back in the priority queue of slots to be used in future iterations of the loop (lines 15 and 16). Finally, the end time of the last task to complete is returned as a proxy to the job execution time.

Figure 8(a) shows an example of modeling the execution of five tasks (\(T_1\)–\(T_5\)) on a node with three available slots, while two blocks (\(B_3\), \(B_4\)) are being prefetched. In this example, each task \(T_i\) is planned to process the corresponding block \(B_i\), and all five blocks are stored on HDDs. Initially, the tasks \(T_1\), \(T_2\), and \(T_5\) are scheduled for execution on the three available slots, because their tier preference score is lower than the score of \(T_3\) and \(T_4\) (as there are pending prefetching operations for the latter two tasks). In this manner, the scheduling of \(T_3\) and \(T_4\) is correctly deferred until slots become available, allowing prefetching to complete. Once \(T_5\) ends and the third slot becomes available, \(T_3\) is scheduled and processes its data from memory. Similarly, \(T_4\) is finally scheduled after \(T_2\) completes execution. Algorithm 5 will return the time 10, as that is the end time of the last task to complete (\(T_4\)).

The final parameter in Algorithm 5 is a delay time parameter, which can be added for delaying the submission of the job, and thus delaying the execution of the first wave of tasks, to give the file system more time to cache the data before the tasks start execution. This new (optional) feature can be very useful in certain scenarios as it can allow some tasks to process prefetched data in time. For example, suppose a fourth slot was available in the example of Figure 8(a). In this case, task \(T_4\) would get scheduled to start at time 3, before the corresponding block completes prefetching. Thus, prefetching \(B_4\) in memory is wasted. However, if the execution of \(T_4\) is delayed by 1 second, then it can start as soon as \(B_4\) completes prefetching and read the data from memory. Whether (and how much) a delay is beneficial for the overall job execution is explored during the data prefetching algorithm presented next.

In this work, data prefetching is formulated as a cost-based optimization problem: Given a job to be run on some input data and cluster resources, find the set of input data blocks to prefetch that will minimize the execution time of the job. The key idea of our approach is to enumerate different sets of input data blocks to prefetch and then simulate the execution of tasks without and with prefetching those blocks, using the models presented in the previous section. During the enumeration, the algorithm keeps track of the prefetching set that led to the lowest job execution time. However, the total number of possible subsets of input data blocks is exponential and thus very inefficient to enumerate. Instead, the proposed enumeration process is based on two important observations. First, disk contention can significantly prolong the time needed for prefetching data into memory, which in turn increases the probability of scheduling tasks before prefetching completes (and thus wasting prefetching). At the same time, it is important to take advantage of the presence of multiple disks across nodes and prefetch multiple blocks in parallel to maximize the number of tasks that can process prefetched data. Therefore, the enumeration tries to generate the largest set of data blocks to prefetch, while each time respecting a limit on the maximum number of blocks to prefetch from each device, which we call the caching degree of parallelism (DoP).

Algorithm 6 shows the approach for selecting a set of input data blocks to prefetch, while respecting a specified maximum caching DoP. The input consists of a list of input data to consider, information about the available cluster resources, and the maximum caching DoP. This algorithm returns the input blocks to prefetch as well as sets the best location for caching each block. For each input data file and for each input block, the algorithm finds the best candidate cache location, defined the location with the lowest storage type and the highest cache remaining space among the locations hosting the block replicas (lines 5–11). The rationale for this decision, which is also used by HDFS and OctopusFS when caching data, is twofold: (1) to maximize the benefits from prefetching (by caching data located in low tiers) and (2) to balance the use of the cache across the cluster nodes. If the candidate cache location respects the maximum caching DoP, then it is set as a source location for caching the block, while the block is added to the list of blocks to prefetch (lines 12–16).

Algorithm 7 outlines the overall data prefetching algorithm. The input consists of a list of input data to process, information about the available cluster resources, and the job configuration. The output is a list of the best blocks to prefetch and a potential delay time for optimizing the job execution time. The first step is to generate the list of tasks for execution based on the input data and the job configuration (line 2). Typically, each task will process one input data block, where each block is replicated across multiple nodes and storage tiers. The execution of tasks is then modeled using Algorithm 5 without prefetching any data to establish a baseline of execution (line 3). Next, the algorithm enters a loop, where each time the maximum caching DoP increments by 1 (to try all relevant maximum caching DoP starting from 1), until a termination condition is met (lines 5 and 6). The loop will terminate when either all input blocks are prefetched or when prefetching any more blocks will not lead to a reduced job execution time. In each iteration, the algorithm selects a set of blocks to prefetch that respects the maximum caching DoP using Algorithm 6, models the prefetching of those blocks using Algorithm 4, and then models the task execution with prefetching using Algorithm 5 (lines 7–9). In addition, if the optional delay time feature is enabled, then the algorithm computes the execution delay time so that all tasks can start execution after prefetching has completed, and then models the task execution with that delay (lines 10–12). Finally, if prefetching with or without delay leads to a lower job execution time, then that particular set of input blocks to prefetch as well as the computed delay time are maintained (lines 13–16) and returned at the end of the algorithm (line 17).

Figure 8 shows an example with two iterations of the main loop of Algorithm 7, when max caching DoP is set to 1 and 2. In the first case, only one block from each HDD device is prefetched (\(B_3\) and \(B_4\)) and the two corresponding tasks (\(T_3\) and \(T_4\)) are able to read the data from memory, speeding up their execution, as well as the overall execution of the job. In the second case, four blocks are prefetched (\(B_2\)–\(B_5\)), two per disk. Since there are only three slots available to run the five tasks, two tasks (namely \(T_2\) and \(T_3\)) will start execution before the corresponding prefetching completes and thus not benefit from it. The other two tasks (\(T_4\) and \(T_5\)) will process prefetched data from memory, but the overall job execution suffers. Prefetching the fifth block when max caching DoP is set to 3 will also be wasteful and will not improve the job execution (not shown in the figure). Note that without prefetching, all five tasks read the data from the HDDs and the total job execution time is 13 seconds. Hence, in this example, prefetching blocks \(B_3\) and \(B_4\) will lead to the lowest job execution time of 10 seconds.

Overall, the Trident Data Prefetcher is used to decide which input data blocks to prefetch into memory for optimizing job execution time. Once the prefetching requests are submitted, the underlying storage system keeps track of both pending and completed prefetch operations. This information is exposed to higher-level platforms, including the Trident Task Scheduler, which will take it into consideration during scheduling, by adjusting the tier preference scores as discussed in Section 4.1. The Trident Resource Scheduler is also informed through the preference map of the extended resource request model presented in Section 3.2. Therefore, instead of trying to predict how a scheduler will behave for making prefetching decisions (like previous works), our approach uses cost modeling for identifying the best input data blocks to prefetch and then uses the schedulers for making appropriate scheduling decisions.

As discussed in Section 3, scheduling based on locality preferences in Hadoop actually takes place at two distinct locations: (1) the RM for allocating containers to applications, and (2) the AM for assigning tasks to the allocated containers. Consequently, the Trident Resource Scheduler is implemented as a pluggable component overriding the scheduling interface provided by Hadoop YARN and is running in the RM for making storage-tier-aware allocation decisions using Algorithm 3. YARN’s resource request model is also extended to include the preference map, as presented in Section 3.2.

Once the MapReduce AM receives a set of allocated containers, it needs to assign tasks to them. The Trident Task Scheduler replaces the default task scheduler in the AM for making optimal storage-tier-aware assignments. In particular, Trident builds a bipartite graph containing map tasks (the only type of tasks with locality preferences in MapReduce) and the allocated containers along with the assignment costs, as described in Section 2.2. In the case of MapReduce, the number of allocated containers will always be less than or equal to the number of tasks. Hence, only Algorithm 2 is implemented in MapReduce for pruning excess tasks when the containers are allocated. Next, Trident employs the Hungarian Algorithm for finding the optimal task assignments on the allocated containers. Regarding reduce tasks (which do not have locality preferences), they are randomly assigned to their allocated containers in the same manner performed by the default task scheduler.

Finally, the data prefetching algorithm discussed in Section 4.2 is implemented by the Trident Data Prefetcher, which runs in the MapReduce Client and is invoked as soon as a MapReduce job is ready to be submitted for execution, for deciding which input data blocks to prefetch into memory. Two more changes were required for applying the Trident methodology. First, Hadoop was modified to propagate the storage tier information from the input file readers to the schedulers, in the same way node locations are propagated. Second, HDFS and OctopusFS were modified to also expose the locations of the pending cache requests, in addition to the locations storing the physical block replicas. Overall, we added 3,425 lines of Java code to Hadoop.

A Spark application executes as a set of independent Executor processes coordinated by the Driver process. Initially, the Driver connects to a cluster manager (either Spark’s Standalone Manager [10], Hadoop YARN [59], or Mesos [36]) and receives resource allocations on cluster nodes that are used for running the Executors. The Driver is responsible for the application’s task scheduling and placement logic, while the Executors are responsible for running the tasks and storing the tasks’ output data over the entire duration of the application.

Internally, an application is divided into jobs, where each job is a directed acyclic graph of stages, and each stage consists of a set of parallel tasks. Whenever a stage S is ready for execution (i.e., its input data are available), the Spark Task Scheduler is responsible for assigning S’s tasks to the available resources (or slots) of the Executors. The default scheduling algorithm is as follows. Given some available slots on Executor E running on some node N, look for a task that needs to process a data partition cached on E, thus creating a process-local assignment. Otherwise, look for a task that needs to process a data block stored on N, thus creating a data-local assignment. Otherwise, make a random assignment if the task has no locality preferences, or a rack-local assignment, or a remote assignment, in that order.

The Trident Task Scheduler is proposed to replace the current task scheduler in the Spark Driver to take advantage of the storage tier information of the processed data. The input to Trident consists of (i) a list of tasks belonging to the same stage along with their preferred locations, and (ii) the list of Executors, each with its available resource set. Given this input, Trident utilizes the pruning Algorithms 1 and 2 to select which resources and tasks to use, builds the bipartite graph as described in Section 2.2, and uses the Hungarian Algorithm for making the optimal task assignments. Similarly to Hadoop, Spark was modified to propagate the storage tier information from the input file readers to the task scheduler. Overall, we added 944 lines of Scala code to Spark.

The Spark execution model creates two additional noteworthy scheduling scenarios that are naturally handled by our graph encoding. First, the preferred location of a task T can be an Executor E containing a cached data partition created during a previous stage execution. Assigning task T to E achieves the best locality possible as it leads to a process-local execution. In this case, we set the tier preference score to zero, thereby guiding Trident in favoring process-local assignments over all other. Second, tasks that will read input from multiple locations during a shuffle (e.g., tasks executing a reduceByKey) have no locality preference. Since the current task scheduler schedules such tasks before making any rack-local (or lower) assignments, we set their assignment cost to a number lower than the network cost \(c_1\) to ensure that Trident has the same behavior. In conclusion, our graph-based formulation can easily generalize to a variety of locality preferences for task assignment.

This part of the evaluation is based only on a MapReduce workload as it is derived from real-world production traces from a 600-node Hadoop cluster deployed at Facebook [16]. With the traces, we used the SWIM tool [56] to generate and replay a realistic and representative workload that preserves the original workload characteristics, including the distribution of input sizes and skewed popularity of data [7]. The workload comprises 1,000 jobs scheduled for execution sporadically over 6 hours and processing 92 GB of input data. Hence, the workload exhibits variability in terms of cluster resource usage over time. When the workload starts execution, 380 files already exist on the file system with a total size of 32 GB. The popularity of these files is skewed, as typically observed in data-intensive workloads [7, 15], with a small fraction of the files accessed very frequently, while the rest are accessed less frequently. In particular, these 380 files are accessed by 660 jobs, with 76.8% of the files accessed once and 4.5% accessed more than 5 times. The remaining 340 jobs are accessing 63 input files generated by other prior jobs that executed on the cluster, with a lower popularity skew: 52.4% of the files are accessed once while 25.4% are accessed more than 5 times. The remaining 947 files generated by the workload have a size of 47 GB and are not accessed by the workload. When using OctopusFS, we enabled its Least Recently Used eviction policy [33] so that later jobs in the workload can take advantage of the memory tier (since the aggregate capacity of the memory tier is 40 GB).

To differentiate the effect of scheduling on different jobs, we split them into six bins based on their input data size. Table 2 shows the distribution of jobs by count, cluster resources they consume, and amount of I/O they generate. As noted in previous studies [7, 15], the jobs exhibit a heavy-tailed distribution of input sizes. Even though small jobs that process \(\lt\)128 MB of data dominate the workload (74.4%), they only account for 25% of the resources consumed and perform only 3.2% of the overall I/O. However, jobs processing over 1 GB of data account for over 54% of resources and over 68% of I/O. More in-depth workload statistics can be found in Reference [33].

We executed the workload on Hadoop over HDFS (without tiering) using the Default and Trident schedulers as well as on Hadoop over OctopusFS using all four schedulers (Default, H-Scheduler, Quartet, and Trident). Data prefetching is disabled and will be investigated later in Section 7.4. Figure 9 shows the data locality rates for all schedulers over the two file systems, broken down according to the job bins. With HDFS and the Default Scheduler, there is a clear increasing trend in the percentage of data-local tasks (note that all data are stored on HDDs) as the job size increases. The achieved data locality is low at 30–40% for small jobs (Bins A and B) for a combination of reasons: (i) the cluster is busy, (ii) these jobs have only a few tasks to run, and (iii) the scheduler considers one node at a time for task assignments. In particular, when a new job is submitted in the cluster, there are on average 3.4 jobs and 8.2 tasks already running and processing (i.e., reading and writing) 1.5 GB of data, while in the worst case, there are 17 jobs and 50 tasks processing 21.0 GB of data. Hence, it is unlikely that any given node will be contained in the tasks’ preferred locations. With increasing job sizes (and number of tasks), there are more opportunities for data-local scheduling and the data locality percentage increases up to 81%. The Trident Scheduler, however, considers all available resources together when building the bipartite graph of tasks and resources, and hence, it is able to achieve almost 100% of data locality for all job sizes.

With OctopusFS, the trend of data-local tasks for the Default Scheduler is the same as with HDFS (see Figure 9). As the Default Scheduler ignores the storage tier, those data-local tasks are (roughly) divided equally into memory-, SSD-, and HDD-local tasks. The H-Scheduler and Quartet have similar or slightly higher overall data-locality rates compared to the Default Scheduler. For small jobs, since there are little opportunities for data-local tasks (for the same three reasons explained above), there is also little chance for doing any meaningful storage-tier-aware task assignments. For bigger jobs, both schedulers are able to make more memory-local assignments, reaching 30-40% of the total tasks and around 50% of the data-local tasks, because the likelihood of finding a task that can be memory-local on a particular node is increased with more tasks. In addition, SSD-local tasks are typically more compared to HDD-local tasks. With OctopusFS, not only is Trident able to reach almost 100% of data locality for all job sizes, it also obtains over 83% of memory-local tasks. In fact, in four of the six bins, Trident is able to achieve over 99% of memory-local tasks, showcasing Trident’s ability to find optimal tasks assignments in terms of both locality and storage tier preferences in a busy cluster.

Figure 10(a) shows the percentage reduction in job completion time compared to using the Default Scheduler over HDFS for each bin (recall Table 2). Using the Trident Scheduler over HDFS improves the overall data-local rates as explained above, which in turn reduces job completion time modestly, up to 13% for large jobs (Bins F). Much better benefits are observed when data are stored in OctopusFS as data are residing in multiple storage tiers, including memory and SSD. Even though the Default Scheduler over OctopusFS does not take into account storage tiers, it still benefits from randomly assigning memory- and SSD-local tasks, and hence, it is able to achieve up to 20% reduction in completion time for large jobs. The storage-tier-aware schedulers are able to increase the benefits further, depending on the job size. Small jobs (Bins A and B) experience only a small improvement (\(\lt\)\(8\%\)) in completion time for all schedulers. This is not surprising, since time spent on I/O is only a small fraction compared to CPU and network overheads. The gains in job completion time increase as the input size increases, while different trends across the schedulers are also observed. In particular, H-Scheduler is able to provide an additional 2%–8% gains over the Default Scheduler, resulting in up to 28% gains for large jobs (Bin F). Quartet offers similar performance, with only 3% higher gains for jobs belonging in bins C and E. Finally, Trident is able to consistently provide the highest reduction in completion time across all job bins, with 14%–37% gains for large jobs, almost double compared to the Default Scheduler over OctopusFS, due to the significantly higher memory locality rates it is able to achieve as observed in Figure 9.

With each memory- and SSD-local access, the cluster efficiency improves as there is more I/O and network bandwidth available for others tasks and jobs. Figure 10(b) shows how this improvement relates to the different job bins. Larger jobs have a higher contribution in efficiency improvement compared to smaller jobs, since they are responsible for performing a larger amount of I/O (recall Table 2). Across different schedulers, the trends for efficiency improvement are similar to the trends for completion time reduction shown in Figure 10(a) and discussed above: Benefits improve with larger jobs and Trident always offers the highest gains. Hence, improvements in efficiency are often accompanied by lower job completion times, doubling the benefits. For example, Trident is able to reduce completion time of large jobs by 37%, while consuming 50% less resources.

Even though Trident achieves on average 91% memory-local tasks for the large jobs compared to 35% of the other two storage-tier-aware schedulers, it only yields about 10% additional reduction in completion time and 16% additional improvement in cluster efficiency over them. The reasons behind this discrepancy are as follows. Job completion time depends on the execution time of all the map tasks, the shuffle stage (which in Hadoop happens as part of the reduce task execution), and the reduce task processing. Increased memory locality will primarily reduce the input I/O time of the map tasks, while it will not impact the CPU processing time of the map and reduce tasks. An interesting observation (not shown in the figures) is that the average execution time of reduce tasks in Hadoop also decreases by up to 37% for large jobs due to the improved cluster efficiency, even though their scheduling is not affected by Trident. The high memory-locality rates achieved by Trident reduce local disk I/O and network congestion, which in turn reduce the time needed for data shuffling between map and reduce tasks. Overall, even though the improvements in completion time and cluster efficiency are not proportional to the improvement in memory locality, Trident is still able to offer the highest improvements compared to the other state-of-the-art approaches.

Key takeaways. Storage-tier-aware scheduling with Trident leads to almost \(100\%\) data locality rates and very high memory-locality rates (\(\gt\)\(83\%\)) across all job sizes, while the second best scheduler is able to achieve only up to \(86\%\) data-locality and \(39\%\) memory-locality rates. In addition, Trident consistently provides the highest benefits in terms of application completion time and cluster efficiency across all scenarios studied.

To further investigate the impact of task scheduling on a variety of workloads exhibiting different characteristics, we used the popular HiBench benchmark v7.1 [37], which provides implementations for various applications on both Hadoop MapReduce and Spark. In total, eight applications were used spanning four categories: micro benchmarks (TeraSort, WordCount), OLAP queries (Aggregation, Join), machine learning (Bayesian Classification, k-means Clustering), and web search (PageRank, NutchIndex) [35]. Table 3 lists the applications along with the physical resource(s) that dominate their execution time. In addition, all workloads were executed using three data scale profiles defined by the benchmark, namely small, large, and huge, which resulted in about 200 MB, 1.5 GB, and 10 GB of input data per application, respectively.

Since the individual workload characteristics (i.e., whether the application is CPU-bound or I/O-bound or network-bound) do not affect data locality rates, we present the aggregate rates for each scale profile in Figure 11. The individual data locality rates per application are very similar to the aggregated ones. The overall trend of data locality rates is similar to the one observed for the Facebook workload: larger jobs exhibit more data-local tasks. However, in these experiments, the rates are much higher, since the cluster is lightly loaded, and thus there are more opportunities for data-local scheduling (note that HiBench runs one application at a time). Hence, the Default Scheduler is able to achieve 66%–89% of data locality instead of 31%–81% in the case of Facebook. Compared to the Default Scheduler, the H-Scheduler and Quartet offer no to little improvement in terms of memory-locality for small and large jobs. The two schedulers are able to achieve good results only when scheduling huge jobs, for which there are a lot of available resources in the cluster, resulting in about 62% memory-local tasks, followed by 19% SSD-local tasks, and 10% HDD-local tasks. Having a lot of available resources in a cluster can improve data locality for huge jobs, because it increases the likelihood of scheduling several data-local (and in the case of H-Scheduler and Quartet, memory-local) tasks on the available resources of any cluster node. This does not necessarily apply to the smaller jobs, because the likelihood of finding a task that can be memory-local on a particular node is reduced. Finally, the Trident Scheduler over OctopusFS is able to achieve 100% data locality with over 96% memory-locality across all three data scales due to the optimality guarantees of the minimum cost maximum matching formulation, demonstrating once again its superior scheduling abilities irrespective of workload size or characteristics.

Figure 12(a) shows the percentage reduction in completion time (compared to the Default Scheduler over HDFS) of the eight HiBench applications run using the large data scale. As expected, I/O intensive applications (i.e., TeraSort, Aggregation, k-means) display the highest benefits across all schedulers, since scheduling more memory-local tasks has a direct impact in reducing both the generated I/O and by extension the overall job execution time. Simply using the Default Scheduler over OctopusFS results in 23%–31% higher performance for these applications, while H-Scheduler increases the benefits to 25%–35%. Interestingly, Quartet offers almost no benefits over the Default Scheduler, mainly because it falls back to delay scheduling when it cannot make any data-local assignments [19]; a strategy that does not increase data locality rates in this setting, and thus, only causes overhead. Finally, Trident is able to significantly boost performance up to 44% (i.e., almost \(2\times\) speedup) due to its 100% memory-locality rates.

The CPU-intensive jobs (i.e., WordCount, Join, Bayes, PageRank) exhibit more modest benefits, since the I/O gains from improved scheduling are overshadowed by the high CPU processing needs. The benefits from Trident over OctopusFS range between 23% and 29%, while they are much lower for the other three schedulers at 8%–23%. Finally, Trident is the only scheduler able to offer any meaningful benefits (28% compared to \raise.17ex\(\scriptstyle \mathtt {\sim }\)6% for the other schedulers) to the shuffle-intensive NutchIndex job, because running 100% data-local tasks frees up the network for the demanding shuffle process.

Figure 12(b) shows the corresponding improvement in cluster efficiency for the large HiBench applications. The efficiency results have the same trends with the reductions in completion times discussed above, but interestingly the magnitude of the gain is higher. The reason is twofold. First, the jobs are executed as a set of parallel tasks. Even if a large fraction of the tasks consume less resources via avoiding disk I/O, the remaining tasks may delay the overall job completion. Second, the job completion time also accounts for CPU processing as well as the output data generation, both of which are independent of the input I/O [33].

The results for the small and huge data scale are similar in trend and omitted due to space constraints. The main difference is the magnitude in gains, which is typically lower for the small scale and higher for the huge scale (compared to the large scale) for all schedulers. The highest reduction in completion time was recorded for the huge Aggregation job using the Trident Scheduler at 57% (i.e., \(2.3\times\) speedup).

In the above experiments, the memory tier of OctopusFS is configured to use 40 GB (i.e., 4 GB per node), which is sufficient for storing 1 replica of the input dataset of each application of the huge HiBench workload. To evaluate the impact of the memory configuration on the benefits provided by Trident, we repeated the experiment with the huge HiBench MapReduce workload but constrained the capacity of the memory tier to 5 GB (i.e., 512 MB per node). With this setup, only about half of the input data can have a replica in the memory tier. Figure 13(a) compares the aggregate data locality rates achieved by all schedulers over OctopusFS for the two memory sizes for the huge HiBench workload. When the memory is constrained to 5 GB, the Trident Scheduler is able to achieve almost 50% memory locality, because it is able to optimally schedule the tasks that are planning to process a block with a replica in memory. Another 48% of tasks are SSD-local, while the remaining tasks are split between HDD-local and rack-local assignments. The other two tier-aware schedulers achieve around 36% of memory locality, which is close to half compared to the 63% of memory locality achieved when all input data reside in memory. The same trend is observed for the Default Scheduler, whose memory locality reduces from 28% to 15%. Hence, the achieved memory locality rate is proportional to the number of input blocks that reside in memory.

Figure 13(b) and (c) respectively show the average percentage reduction in completion time and percentage improvement in cluster efficiency compared to the baseline for the huge HiBench workload for the two memory settings. As expected, the benefits achieved by all schedulers is lower for both metrics due to the reduced memory locality rates, but not by the same amount. In particular, the reduction in completion time of H-Scheduler and Quartet is reduced to half in the memory-constraint scenario, whereas for Trident it decreases from 47% to 33%. When no more tasks can be scheduled in a memory-local way, Trident shifts the assignment of tasks in an SSD-local way (more than the other schedulers), leading to good benefits for both application performance and cluster efficiency.

Key takeaways. These performance results clearly demonstrate that Trident can achieve very high memory-locality rates (\(\gt\)\(95\%\)) irrespective of application size or characteristics. The benefits in terms of completion time and cluster efficiency are typically higher for I/O- and network-intensive applications as well as for larger jobs.

The evaluation with the HiBench workloads was repeated on Spark in the same manner as on Hadoop (described in Section 7.2), with the exception of NutchIndex, which is not implemented for Spark. We used Spark’s Standalone Cluster Manager for allocating resources across applications, which is widely used in practice [19]. Each application received one Executor process on each worker node, while the Driver process was executed on the Master node.

The overall data locality rates for all schedulers for the HiBench Spark workload are shown in Figure 14. The first key observation is that, unlike Hadoop, the Spark Default Scheduler is able to achieve over 94% data locality across all data scales. There are two reasons explaining this behavior. First, each application has available resources on all nodes, and hence, can selectively choose which ones to use for the task assignments (especially for smaller jobs), unlike MapReduce that gets resources on some nodes based on containers allocated from YARN. Second, the Default Scheduler has a built-in load balancing feature that iterates the available resources on each node one Executor slot at a time, which increases the opportunities for data-local assignments (or memory-local in the case of H-Scheduler and Quartet). These features are shared by the H-Scheduler and Quartet as well and are thus are able to achieve high memory locality rates of over 71% and 84% for large and huge applications, respectively. Trident, however, is able to achieve 100% memory locality for both small and large applications, as well as 96% memory locality for huge applications, significantly outperforming all other schedulers. Finally, note that process-local tasks are all assigned in a separate process, before the other tasks are assigned; hence, the percentage of process-local tasks is the same for all schedulers over both file systems.

In terms of reduction in completion time, the overall trends are similar as in the case of running the applications on Hadoop, and are shown in Figure 15 for the large and huge data scales. In particular, the H-Scheduler and Quartet are able to offer good performance improvements over the Default Scheduler, because they are able to exploit the storage tier information, but are still outperformed by Trident in all cases. The magnitude of gains for the large iterative applications (i.e., Bayes, k-means, and PageRank) are lower for Spark compared to Hadoop, because Spark will cache the output data from the first iteration in memory and then use process-local tasks for the following iterations. Hence, the gains from memory-local task assignments only impact the first iteration. Even then, in the case of huge Bayes and k-means, Trident is able to speedup their first iteration by \(4\times\), leading to an overall application speedup of over \(2\times\). Spark’s PageRank does not enjoy such benefits, because its first iteration is very CPU intensive, thus limiting the I/O gains from memory-locality. Another interesting observation is that, unlike with Hadoop, Quartet is able to outperform H-Scheduler in Spark by 4% on average in most cases, because its delay scheduling approach is actually able to improve memory-locality rates by 2%–5%. Finally, in the case of the huge workload, Trident is able to significantly improve performance for most applications, reaching up to 66% reduction in completion time, i.e., \(3\times\) speedup.

Key takeaways. The benefits in terms of improved locality, reduction in completion time, and reduction in cluster efficiency provided by Trident to Spark workloads and clusters are very similar to the ones provided for Hadoop.

In this section, we investigate the impact of data prefetching in addition to scheduling over tiered storage. We experiment with two flavors of the Trident Data Prefetcher, one that does not allow any delays in the execution of tasks (denoted as Trident-D) and one that does (denoted as Trident+D), as discussed in Section 4.2. For comparison purposes, we implemented a prefetcher (called PrefetchAll) that will issue prefetching requests for all data needed by each job upon its submission. Finally, we also implemented an Oracle prefetcher for prefetching the data needed before each job submission, so that one input block replica is always available in memory when needed. The Oracle prefetcher cannot exist in practice but we implemented one to establish an upper bound on the potential benefits provided by any prefetcher.

For these experiments, we used both the Facebook workload and the HiBench MapReduce workload with the huge data scale. The input data are initially stored using three replicas on HDDs, while prefetching requests cache one replica into memory. We repeated the experiments using HDFS and OctopusFS as the underlying storage system but the results were very similar. Thus, we only show the results for OctopusFS. Finally, to demonstrate the impact (and necessity) of a storage-tier-aware scheduler when prefetching is enabled, we also run the PrefetchAll prefetcher with the Default Scheduler. In all other cases, we use the Trident Scheduler.

Figure 16 shows the percentage of input blocks accessed from memory and HDDs for each scheduler-prefetcher combination. For the Facebook workload, we break down the results to the six job sizes for Bins A–F (recall Table 2), while for the HiBench workload we present the aggregated results from the eight applications (see Table 3). Since the Trident Scheduler is able to achieve near \(100\%\) data-locality in all scenarios, and to keep the figure readable, we do not differentiate between node-local and rack-local block accesses. The figure also shows the percentage of blocks that were prefetched but not read from memory by the tasks, i.e., prefetching was wasted. This figure enables direct comparisons across multiple dimensions, showcasing the impact of scheduler, prefetcher, and their combination on both memory locality and wasted prefetch operations.

Our first observation from Figure 16 confirms that prefetching data without a proper task scheduler in place will result in many wasted prefetching requests. For example, in the case of small Facebook jobs (Bin A), prefetching all input data and using the Default Scheduler, which does not take storage tiers into account, leads to \(50\%\) of memory accesses, thereby wasting the prefetching of the other \(50\%\). However, using the Trident Scheduler leads to \(100\%\) of memory accesses when prefetching all data, revealing that prefetching completes before the tasks begin execution (due to the small input size). As the job size increases, prefetching all input data leads to I/O contention when caching data from the HDDs into memory. This causes delays in prefetching and many tasks are scheduled for execution before prefetching completes. As a result, the percentage of wasted block prefetches increases with job size when the PrefetchAll is used. For the larger jobs (Bins E–F), over \(84\%\) of prefetches are wasted with either scheduler. Therefore, prefetching all input data is not a viable strategy for improving cluster performance.

In the case of small jobs, the Trident Data Prefetcher (with and without delay) correctly prefetched all data and achieved \(100\%\) memory accesses. However, as the job size increases, the Trident Data Prefetcher becomes increasingly selective and only prefetches a small portion of the overall data; the portion that avoids disk contention and ensures that the prefetched data will be read by the tasks. As a result, the percentageof memory accesses decreases with increasing job size but only a very small fraction, if any, of the prefetch requests are wasted (less than \(4\%\)). When the delay feature is enabled, the Trident+D Prefetcher is able to achieve higher percentages of memory accesses for the mid-sized jobs (Bin B-E) compared to Trident-D, as also shown in Table 4. For instance, \(27.6\%\) of the jobs belonging to Bin D are delayed on average by 3.8 seconds with a max delay of 6.1 seconds. As a result, the percentage of memory accesses increases from \(22\%\) with Trident-D to \(64\%\) with Trident+D, leading to an overall \(12\%\) decrease in the average job completion time (including the delay). For smaller jobs, Trident+D will delay a smaller percentage of jobs for a smaller duration, because prefetching completes quickly, making delays unnecessary. For larger jobs, Trident+D will also delay a small percentage of jobs, because delaying the job to wait for prefetching to complete will typically eliminate the benefits of reading data from memory. In fact, for Bin F, Trident+D will choose not to delay any job and behaves identically to Trident-D. In addition to reducing the mean job completion time, Trident+D is also able to reduce the 99% tail job completion time by up to 13% for mid-sized jobs (Bin D) compared to Trident-D because of the large percentage of jobs (\(27.6\%\)) that are able to benefit from prefetching with delay. Finally, as expected, the Oracle Prefetcher leads to very high memory access percentages of over \(96\%\) for the Facebook workload and \(87\%\) for the HiBench workload, with small corresponding waste.

Figure 17(a) and (b) show the percentage reduction in completion time and percentage improvement in cluster efficiency, respectively, for the various scheduling and prefetching combinations compared to the Default Scheduler without prefetching over OctopusFS. Prefetching all data when using the Default Scheduler yields almost no benefits with regards to application performance (despite the increased memory access rates), while it will hurt cluster efficiency due to the large amount of wasted prefetches. Prefetching all data when using the Trident Scheduler leads to some modest reduction in completion time (4–11\(\%\)) due to the increased memory-local tasks, but these benefits come in the expense of reduced cluster efficiency due to the wasted prefetches. The Trident Data Prefetcher, with its good memory access rates and minimal waste, always leads to higher reductions in completion times (up to \(17\%\)), while at the same time achieving high cluster efficiency gains (up to \(31\%\)). Trident+D, with its targeted use of delaying some task executions, it is able to provide some additional benefits compared to Trident-D, up to \(3.7\%\) reduction in mean completion time and \(8.5\%\) improvement in cluster efficiency.

Finally, we compare the scenario of the Oracle Prefetcher against the Trident Data Prefetcher. Recall that in the former case, jobs begin execution with their input data already cached in memory, while in the later case, no data are initially in memory. Hence, the Oracle case represents the theoretically best benefits achievable by any prefetcher. For large jobs, it is not possible to prefetch all data in memory while the job is running and achieve good memory access rates. Hence, there is some observed difference between the Oracle and the Trident experiments, up to \(8\%\) in terms of job completion time. For smaller jobs, however, the achieved benefits of Trident are very close to the ones of the Oracle Prefetcher (less than \(3\%\) difference), showcasing the near-optimal decisions made by the Trident Data Prefetcher.

Key takeaways. From the results, it becomes clear that the Trident Data Prefetcher, with the algorithms and modeling presented in Section 4, is able to navigate the various tradeoffs effectively and balance the amount of data to prefetch as well as specify an appropriate delay (if any), to maximize the benefits from prefetching data over tiered storage.

We begin the investigation of overheads by evaluating Trident’s scheduling time as we vary the number of tasks n ready for execution and the number of cluster nodes r with available resources. For this purpose, we instantiate a Spark Manager and register r virtual nodes, each with one Executor. Next, we submit a Spark application with one stage of n tasks. Each task has a list of 3 preferred locations (i.e., \(\langle\)node, tier\(\rangle\) pairs) in random nodes and tiers across the cluster. Finally, we measure the actual time needed by Trident to make all possible task assignments. This time also includes updating all relevant internal data structures maintained by the Spark Driver.

Figure 18 shows the scheduling times as we vary both n and r between 8 and 1,024 (note the logarithmic scale of both axes), when our two pruning algorithms are either disabled or enabled. With pruning, as long as one of the two dimensions (i.e., tasks or nodes) is small, the scheduling time is very low and grows linearly. For example, with up to 64 tasks, the scheduling time with pruning is below 2 ms regardless the cluster size, whereas it can reach 20 ms without pruning for 1,024 nodes (i.e., there is an order of magnitude reduction). Similarly, large jobs (\(n\ge 256\)) get scheduled quickly in under 3ms in small clusters (\(r\le 32\)), whereas scheduling time can reach 54ms without pruning. The scheduling time increases non-linearly only when both dimensions are high, since pruning cannot help and the main algorithm’s complexity is \(O(\mbox{min}(n, r)^3)\). However, even in the extreme case of scheduling 1,024 tasks on a 1,024-node cluster, the scheduling time is only 240 ms. More importantly, this overhead is incurred by the Spark Driver (or the MapReduce Application Master in Hadoop) and not the cluster, and is minuscule compared to both the total execution time of such a large job and the potential performance gains from Trident’s scheduling abilities.

We repeated this experiment in Hadoop and the scheduling times in MapReduce AM are very similar to the ones observed for Spark. However, Trident’s scheduling times in YARN’s Resource Manager are much lower, as they are governed by Algorithm 3 with complexity \(O(k lg(k))\), where k is the number of requested containers. Specifically, in the case of 1,024 tasks \(\times\) 1,024 nodes, Trident’s scheduling time is 61 ms as opposed to 60 ms for FIFO and 162 ms for Capacity (extra time due to updating queue statistics after each assignment), highlighting the negligible overheads induced by our scheduling approach.

Finally, we evaluated the time needed for the Trident Data Prefetcher to select which blocks to prefetch, per Algorithm 7. For this purpose, we varied the number of nodes in the cluster (\(4-1,\!024\)), the number of input data blocks (\(4-1,024\)), as well as the number of storage devices per node (\(3-12\)), since our prefetching modeling works at the level of individual storage devices (recall Algorithm 4). In the majority of the tested scenarios, the prefetching algorithm run in less that 2 ms, while in the worst case scenario of 1,024 input data blocks \(\times\) 1024 nodes \(\times\) 12 storage devices per node, it run in 20 ms, showcasing the high efficiency of the Trident Data Prefetcher.

Key takeaways. The very low overheads observed for task scheduling (average: 13.6 ms; max: 240 ms), resource scheduling (average: 9.5 ms; max: 61 ms), and data prefetching (average: 3.2 ms; max: 20 ms) reveal the practicality of the Trident methodology, even when scheduling very large jobs to very large clusters.