Read–Write Behavior

The end of the commit phase marks the switch to the new shard key. Until this point, all queries with the old shard key were routed to the mapped server and all queries with the new shard key were multicast to all the servers (which is normal MongoDB behavior). After the commit phase, a query with the new shard key is routed to the appropriate server (the new primary). Queries that do not use the new shard key are handled with a multicast, which again is normal MongoDB behavior.

Reads in MongoDB offer per-key sequential consistency. Morphus is designed so that it continues to offer the same consistency model for data undergoing migration.

Greedy Assignment

The greedy approach considers each new chunk independently. For each new chunk NC_i, the approach evaluates all the N servers. For each server S_j, it calculates the number of data items of chunk NC_i that are already present in old chunks at server S_j. The approach then allocates each new chunk NC_i to the server S_j that has the maximum value of , that is, . As chunks are considered independently, the algorithm produces the same output irrespective of the order in which chunks are considered by it.

To illustrate the greedy scheme in action, Figure 6.2 provides two examples for the shard key change operation. In each example, the database has three old chunks, OC₁–OC₃, each of which contains three data items. For each data item, we show the old shard key K_o and the new shard key K_n (both in the range 1–9). The new configuration splits the new key range evenly across the three chunks, shown as NC₁–NC₃.

In Figure 6.2a, the old chunks are spread evenly across servers S₁–S₃. The edge weights in the bipartite graph show the number of data items of NC_i that are local at S_j, that is, values. Thick lines show the greedy assignment.

However, the greedy approach may produce an unbalanced chunk assignment for skewed bipartite graphs, as in Figure 6.2b. While the greedy approach minimizes network transfer volume, it assigns new chunks NC₂ and NC₃ to server S₁, while leaving server S₃ empty.

Load Balance via Bipartite Matching

Load balancing of chunks across servers is important for several reasons. First, it improves read/write latencies for clients by spreading data and queries across more servers. Second, it reduces read/write bottlenecks. Finally, it reduces the tail of the reconfiguration time by preventing the allocation of too many chunks to any one server.

Our second strategy achieves load balance by capping the number of new chunks allocated to each server. With m new chunks, this per-server cap is chunks. We then create a bipartite graph with two sets of vertices, top and bottom. The top set consists of vertices for each of the N servers in the system; the vertices for server are denoted by . The bottom set of vertices consists of the new chunks. All edges between a top vertex and a bottom vertex NC_i have an edge cost equal to , that is, the number of data items that will move to server if new chunk NC_i is allocated to it.

Assigning new chunks to servers in order to minimize data transfer volume now becomes a bipartite matching problem. Thus, we find the minimum weight matching by using the classical Hungarian algorithm [106]. The complexity of this algorithm is , where chunks. This reduces to O(m³). The greedy strategy becomes a special case of this algorithm with .

Figure 6.2b shows the outcome of the bipartite matching algorithm with dotted lines in the graph. While it incurs the same overall cost as the greedy approach, it provides the benefit of a load-balanced new configuration, wherein each server is allocated exactly one new chunk.

While we focus on the shard key change, this technique can also be used for other reconfigurations, such as changing shard size, or cluster scale-out and scale-in. The bipartite graph would be drawn appropriately (depending on the reconfiguration operation) and the same matching algorithm used. For the purpose of concreteness, the rest of this section of the chapter focuses on shard key change.

Finally, we have used data size (number of key-value pairs) as the main cost metric. Instead, we could use traffic to key-value pairs as the cost metric and derive edge weights in the bipartite graph (Figure 6.2) from these traffic estimates. The Hungarian approach on this new graph would balance out traffic load, while trading off optimality. Further exploration of this variant is beyond our scope here.

Setup

We used the data set of Amazon reviews as our default collection [110]. Each data item had 10 fields. We chose product ID as the old shard key and userID as the new shard key; update operations used these two fields and a price field. Our default database size was 1 GB. (We later show scalability with data size.)

The default Morphus cluster used 10 machines, which included one mongos (front end) and three replica sets, each containing a primary and two secondaries. There were three config servers, each of which was colocated on a physical machine with a replica set primary; this is an allowed MongoDB installation. Each physical machine was a d710 Emulab node [111] with a 2.4 GHz processor, four cores, 12 GB RAM, two hard disks of 250 GB and 500 GB, and 64-bit CentOS 5.5, connected to a 100 Mbps LAN switch.

We implemented a custom workload generator that injects YCSB-like workloads via MongoDB's pymongo interface. Our default injection rate was 100 ops/s with 40% reads, 40% updates, and 20% inserts. To model realistic key access patterns, we selected keys for each operation via one of three YCSB-like [112] distributions: (i) Uniform (default), (ii) Zipf, and (iii) Latest. For the Zipf and Latest distributions, we employed a shape parameter α = 1.5. The workload generator ran on a dedicated pc3000 node in Emulab that ran a 3 GHz processor, 2 GB RAM, two 146 GB SCSI disks, and 64-bit Ubuntu 12.04 LTS.

Morphus was implemented in about 4000 lines of C++ code, which is publicly available at http://dprg.cs.uiuc.edu/downloads. Each plotted data point is an average of at least three experimental trials, shown along with standard deviation bars. Section 6.4.2.4 outlined two algorithms for the shard key change reconfiguration: Hungarian and greedy. We implemented both in Morphus, and call them variants Morphus-H and Morphus-G, respectively.

While we present only selected experimental results in this chapter, we refer the reader to Ref. [102] for extensive information on experiments and evaluation of our system.

Effect on Reads and Writes

A key goal of Morphus is to ensure the availability of the database during reconfiguration. To evaluate its success, we generated read and write requests and measured their latency while a reconfiguration was in progress. We used Morphus-G with a chunk-based migration scheme. We ran separate experiments for all the key access distributions and also for a read-only workload.

Table 6.1 lists the percentages of read and write requests that succeeded during reconfiguration. The number of writes that failed is low: For the Uniform and Zipf workloads, fewer than 2% of writes failed. We observe that many of the failed writes occurred during one of the write-throttling periods. Recall from Section 6.4.2.3 that the number of write-throttling periods is the same as the replica set size, with one throttle period at the end of each reconfiguration round. The Latest workload has a slightly higher failure rate, since if an attempt has been made to write a particular key, that increases the likelihood that in the near future there will be another attempt to write or read that same key. Still, the write failure rate of 3.2% and the read failure rate of 2.8% are reasonably low.

Overall, the availability numbers are higher, at 99% to 99.9% for Uniform and Zipf workloads, which is comparable to the numbers for a scenario with no insertions. We conclude that unless there is temporal and spatial (key-wise) correlation between writes and reads (i.e., Latest workloads), the read latency is not affected much by concurrent writes. When there is correlation, Morphus mildly reduces the offered availability.

Going further, we plot in Figure 6.3a the CDF of read latencies for the four settings, and for a situation in which there was no reconfiguration (Uniform workload). Note that the horizontal axis is logarithmic scale. We consider latencies only for successful reads. We observe that the 96th percentile latencies for all workloads are within a range of 2 ms. The median (50th percentile) latency for No Reconfiguration is 1.4 ms, and this median holds for both the Read only (No Write) and Uniform workloads. The medians for the Zipf and Latest workloads are lower, at 0.95 ms. This lowered latency has two causes: caching at the mongod servers for the frequently accessed keys, and, in the case of Latest, the lower percentage of successful reads. In Figure 6.3b, we plot the corresponding CDF for write latencies. The median for writes when there is no reconfiguration (Uniform workload) is similar to that of the other distributions.

We conclude that under reconfiguration, the read and write availability provided by Morphus is high (close to 99%), while latencies of successful writes degrade only mildly compared to those observed when there is no reconfiguration in progress.

Effect of Network Awareness

First, Figure 6.4a shows the length of the execution phase (when a 500 MB Amazon collection was used) for two hierarchical topologies and five migration strategies. The topologies were (i) homogeneous, in which nine servers were distributed evenly across three racks; and (ii) heterogeneous, in which three racks contained six, two, and one servers, respectively. The switches were Emulab pc3000 nodes and all links were 100 Mbps. The inter-rack and intra-rack latencies were 2 and 1 ms, respectively. The five strategies were (i) fixed sharing, with one socket assigned to each destination node; (ii) a chunk-based approach (see Section 6.4.2.3); (iii) Orchestra [21] with ; (iv) WFS with (see Section 6.4.2.5); and (v) WFS with .

We observed that in the homogeneous clusters, the WFS strategy with was 30% faster than fixed sharing and 20% faster than the chunk-based strategy. Compared to Orchestra, which weights flow only by their data size, WFS with does 9% better, because it takes the network into account as well. Increasing K from 21 to 28 improves completion time in the homogeneous cluster, but causes degradation in the heterogeneous cluster. The reason is that a higher K results in more TCP connections, and at , this begins to cause congestion at the rack switch of six servers.

Second, Figure 6.4b shows that Morphus's network-aware WFS strategy has a shorter tail and finishes earlier. Network awareness lowers the median chunk finish time by around 20% in both the homogeneous and heterogeneous networks.

We conclude that the WFS strategy improves performance compared to existing approaches, and K should be chosen to be as high as possible without leading to congestion.

Large-Scale Experiment

In this experiment, we increased data and cluster size simultaneously such that the amount of data per replica set was constant. We ran this experiment on Google Cloud [113]. We used n1-standard-4 VMs, each with four virtual CPUs and 15 GB of memory. The disk capacity was 1 GB, and the VMs were running Debian 7. We generated a synthetic data set by randomly dispersing data items among new chunks. Morphus-H was used for reconfiguration with the WFS migration scheme, and K = the number of old chunks.

Figure 6.5 shows a sublinear increase in reconfiguration time as data size and cluster size increased. Note that the x-axis uses a log scale. In the execution phase, all replica sets communicated among themselves to migrate data. As the number of replica sets increased with cluster size, the total number of connections increased, leading to network congestion. Thus, the execution phase took longer.

The amount of data per replica set affects reconfiguration time superlinearly. On the other hand, cluster size has a sublinear impact. In this experiment, the latter dominated, as the amount of data per replica set was constant.

Isolate Phase

In this phase, the initiator node, in which the reconfiguration command is run, creates a new (and empty) column family (i.e., database table), denoted by Reconfigured CF (column family). It does so using a schema derived from the Original CF, except it uses the desired key as the new primary key. The Reconfigured CF enables reconfiguration to happen in the background while the Original CF continues to serve reads and writes using the old reconfiguration. We also record the timestamp of the last operation before the Reconfigured CF is created so that all operations that arrive while the execute phase is running can be applied later in the recovery phase.

Execute Phase

The initiator node notifies all other nodes to start copying data from the Original CF to the Reconfigured CF. Read and write requests from clients continue to be served normally during this phase. At each node, Parqua iterates through all entries for which it is responsible and sends them to the appropriate new destination nodes. The destination node for an entry is determined by (i) hashing the new primary key-value on the hash ring, and (ii) using the replica number associated with the entry. Key-value pairs are transferred between corresponding nodes that have matching replica numbers in the old configuration and the new configuration.

For example, in the execute phase of Figure 6.6, the entry with the old primary key “1” and the new primary key “10” has a replica number of 1 at node A, 2 at B, and 3 at C. In this example, after the primary key is changed, the new position of the entry on the ring is between nodes C and D, where nodes D, E, and F are replica numbers 1, 2, and 3, respectively. Thus, in the execute phase, the said entry in node A is sent to node D, and similarly the entry in B is sent to E, and from C to F.

Commit Phase

After the execute phase, the Reconfigured CF has the new configuration, and the entries from the Original CF have been copied to the Reconfigured CF. Now, Parqua atomically swaps both the schema and the SSTables between the Original CF and the Reconfigured CF. The write requests are locked in this phase, while reads still continue to be served. To implement the SSTable swap, we leverage the fact that SSTables are maintained as files on disk, stored in a directory named after the column family. Therefore, we move SSTable files from one directory to another. This does not cause disk I/O, as we update the inodes only when moving files.

At the end of the commit phase, the write lock is released at each node. At this point, all client-facing requests are processed according to the new configuration. In our case, the new primary key is now in effect, and the read requests must use the new primary key.

Recovery Phase

During this phase, the system catches up with the recent writes that were not transferred to the Reconfigured CF in the execute phase. Read/write requests are processed normally, with the difference that until the recovery is done, the read requests may return stale results.⁶ At each node, Parqua iterates through the SSTables of the Original CF to recover the entries that were written during the reconfiguration. We limit the amount of disk accesses required for recovery by iterating only the SSTables that were created after the reconfiguration started. The iterated entries are routed to appropriate destinations in the same way as in the execute phase.

Since all writes in Cassandra carry a timestamp [118], Parqua can ensure that the recovery of an entry does not overshadow newer updates, thus guaranteeing the eventual consistency.

Setup

We used the Yahoo! Cloud Service Benchmark (YCSB) [112] to generate the data set and used the Uniform, Zipfian, and Latest key access distributions to generate CRUD workloads. Our default database size was 10 GB in all experiments. The operations consisted of 40% reads, 40% updates, and 20% inserts. Again, we present only selected experimental results in this chapter, and refer the reader to Ref. [114] for extensive information on experiments and evaluation of our system.

Availability

In this experiment, we measured the availability of our system during reconfiguration, shown in Table 6.2. The slight degradation in availability was due to the rejection of writes in the commit phase. The total duration of the unavailability was only a few seconds, which is orders of magnitude better than the current state of the art.

The lowest availability was observed for the Latest distribution. The reason is that YCSB does not wait for the database to acknowledge an insert of a key. Because of the temporal nature of the distribution, as keys are further read and updated, the operations fail because the inserts are still in progress.

Effect on Read Latency

Figure 6.7a shows the CDF of read latencies under various workloads while reconfiguration is being executed. As a baseline, we also plot the CDF of read latency when no reconfiguration is being run using the Uniform key access distribution. We plot the latencies of successful reads only.

The median (50th percentile) latencies for the read-only workload and the baseline are similar because they both use Uniform distribution. Under reconfiguration, 20% of the reads take longer. With writes in the workload, the observed latencies for the Uniform curve are higher overall.

Compared to other workloads, Latest had the smallest median latency. Because of that workload's temporal nature, recently inserted keys were present in Memtables, which is a data structure maintained in memory. As a result, reads were faster than in other distributions that require disk accesses.

Overall, Parqua affects median read latency minimally across all the distributions. Our observations for write latency are similar. We refer the reader to our technical report for more detail in Ref. [119].

Scalability

Next, we measured how well Parqua scales with (i) database size, (ii) cluster size, (iii) operation injection rate, and (iv) replication factor. For lack of space, we omit the plots for the last two experiments and refer the reader our technical report for them [119]. To evaluate our system's scalability, we measured the total reconfiguration times along with the time spent in each phase. We did not inject operations for the experiments presented next.

Figure 6.7b depicts the reconfiguration times as the database size is increased up to 30 GB. Since we used a replication factor (number of copies of the same entry across the cluster) of 3 for fault tolerance, 30 GB of data implies 90 GB of total data in the database. In this plot, we observe that the total reconfiguration time scales linearly with database size. The bulk of the reconfiguration time is spent in transferring data in the execute phase.

In Figure 6.7c, we observe that the reconfiguration time decreases as the number of Cassandra peers increases. The decrease occurs because as the number of machines increases, the same amount of data, divided into smaller chunks, gets transferred by a larger number of peers. Again, the execute phase dominated the reconfiguration time.

Effective Throughput Percentage

To estimate the impact of each operator on the application throughput, Stela uses a new metric we developed called ETP. An operator's ETP is defined as the percentage of the final throughput that would be affected if the operator's processing speed were changed.

The ETP of an operator o is computed as

Here, Throughput_{EffectiveReachableSinks} denotes the sum of the throughputs of all sinks reachable from o by at least one uncongested path, that is, a path consisting only of operators that are not classified as congested. Throughput_workflow denotes the sum of the throughputs of all sink operators of the entire application. The algorithm that calculates ETPs does a depth-first search throughout the application DAG and calculates ETPs via a postorder traversal. ProcessingRateMap stores the processing rates of all operators. Note that if an operator o has multiple parents, then the effect of o's ETP is the same at each of its parents (i.e., it is replicated, not split).

ETP Calculation Example and Intuition

In Figure 6.8, we illustrate the ETP calculation with an example application.⁷ The processing rate of each operator is shown. The congested operators, that is, operators 1, 3, 4, and 6, are shaded. The total throughput of the workflow is calculated with Throughput_workflow = 5000 tuples/s as the sum of throughputs of sink operators 4, 7, 8, 9, and 10.

Let us calculate the ETP of operator 3. Its reachable sink operators are 7, 8, 9, and 10. Of these, only 7 and 8 are considered to be the “effectively” reachable sink operators, as they are both reachable via an uncongested path. Thus, increasing the speed of operator 3 will improve the throughput of operators 7 and 8. However, operator 6 is not effectively reachable from operator 3, because operator 6 is already congested; thus, increasing operator 3's resources will only increase operator 6's input rate and make operator 6 further congested, without improving its processing rate. Thus, we ignore the subtree of operator 6 when calculating 3's ETP. The ETP of operator 3 is ETP₃ = (1000 + 1000)/5000 = 40%.

Similarly, for operator 1, the sink operators 4, 7, 8, 9, and 10 are reachable, but none of them are reachable via an uncongested path. Thus, the ETP of operator 1 is 0. Likewise, we can calculate the ETP of operator 4 as 40% and the ETP of operator 6 as 20%. Therefore, the priority order in which Stela will assign resources to these operators is 3, 4, 6, 1.

Fault Tolerance

When no scaling is occurring, failures are handled the same way as in Storm, that is, Stela inherits Storm's fault tolerance. If a failure occurs during a scaling operation, Stela's scaling will need to be aborted and restarted. If the scaling is already committed, failures are handled as in Storm.

Contiguous Vertex Repartitioning (CVR)

Our first technique assumes that the hashed vertex space is divided into as many partitions as there are servers, and each server is assigned one partition. Partitions are equisized in order to accomplish load balancing. The top of Figure 6.14a shows an example graph containing 100 vertices, split across 4 servers. The vertex sequence (i.e., V_i) is random but consistent due to our use of consistent hashing and is split into four equisized partitions, which are then assigned to servers S₁–S₄ sequentially.

Upon a scale-out/scale-in operation, the key problem we need to solve is, how do we assign the (new) equisized partitions to servers (one partition per server), such that network traffic volume is minimized? For instance, the bottom of Figure 6.14 shows the problem when scaling out from four to five servers. To solve this problem, we now (i) show how to reduce the problem to one of graph matching, and (ii) propose an efficient heuristic.

When we scale-out/scale-in, we repartition the vertex sequence into equisized partitions. Assigning these new partitions in an arbitrary fashion to servers may be suboptimal and involve transfer of large amounts of vertex data across the network. For instance, in the bottom of Figure 6.14a, we scale-out by adding one server, resulting in five new partitions. Merely adding the new server to the end of the server sequence and assigning partitions to servers in that order results in movement of 50 total vertices. On the other hand, Figure 6.14b shows the optimal solution for this example, wherein adding the new server in the middle of the partition sequence results in movement of only 30 vertices.

To achieve the optimal solution, we consider the scale-out problem formally. (The solution for scale-in is analogous and excluded for brevity.) Let the cluster initially have N servers S₁,…, S_N. With a graph of V vertices, the initial size of each partition is , where 1 ≤ i ≤ N. Each jth vertex ID is hashed, and then the resulting value is used to assign the vertex to partition where . If we add k servers to this cluster, the size of each new partition becomes . We label these new partitions , 1 ≤ i ≤ N + k, and assign each jth vertex, as usual, to a new partition by first hashing the vertex ID and using the resulting hash to partition where .

Next, we create a bipartite graph B, which contains (i) a left set of vertices, with one vertex per new partition , and (ii) a right set of vertices, with one vertex per server . The left and right sets each contain (N + k) vertices. The result is a complete bipartite graph, with the edge joining a partition and a server associated with a cost. The cost is equal to the number of vertices that must be transferred over the network if partition is assigned to server after scale-out. In other words, the cost is equal to .

The problem of minimizing network transfer volume now reduces to that of finding a minimum-cost perfect matching in B. This is a well-studied problem, and an optimal solution can be obtained by using the Hungarian algorithm [128]. However, the Hungarian algorithm has O(N³) complexity [129], which may be prohibitive for large clusters.

As a result, we propose a greedy algorithm that iterates sequentially through S₁,…, S_N, in that order. ⁹ For each server , the algorithm considers the new partitions with which it has a nonzero overlap; because of the contiguity of partitions, there are only O(1) such partitions. Among these partitions, the one with the largest number of overlapping vertices with is assigned to server . Because of the linear order of traversal, when is considered, is guaranteed to have at least one (overlapping) candidate position. This makes the greedy algorithm run efficiently in O(N). For example, in Figure 6.14b, to determine the new partition for , we need to consider only two partitions, and ; next, we need to consider partitions and , and so on.

Ring-Based Vertex Repartitioning (RVR)

In this technique, we assume an underlying hash-based partitioning that leverages Chord-style consistent hashing [127]. To maintain load balance, servers are not hashed directly to the ring; instead (as in Cassandra [2] and Riak [115]), we assume that each server is assigned an equisized segment of the ring. Specifically, a server with ID n_i is responsible for vertices hashed in the interval , where is n_i's predecessor.

Under that assumption, performing a scale-out/scale-in operation is straightforward: a joining server splits a segment with its successor, while a leaving server gives up its segment to its successor. For instance, in a scale-out operation involving one server, the affected server receives its set of vertices from its successor in the ring, that is, a server n_i takes the set of vertices from its successor . Scale-in operations occur symmetrically: a leaving server n_i migrates its vertex set to its successor , which is then responsible for the set of vertices in .

More generally, we can state that a scale-out/scale-in operation that involves simultaneous addition or removal of k servers affects at most k existing servers. If some of the joining or leaving servers have segments that are adjacent, the number of servers affected would be smaller than k.

While the technique is minimally invasive to existing servers and the ongoing graph computation, it may result in load imbalance. We can mitigate load imbalance for the scale-out case by choosing intelligently the point on the ring to which the new server(s) should be added. For the scale-in case, we can intelligently decide which server(s) to remove.

Consider a cluster with N servers, each with vertices. If we use CVR to add m × N servers or remove servers (for m ≥ 1), then the optimal position of servers to be added or removed is same as their position with RVR.

Executing Migration

LFGraph uses a publish–subscribe mechanism. Before the iterations start, each server subscribes to in-neighbors of the vertices hosted by the server. Based on these subscriptions, each server builds a publish list for every other server in the cluster. After each iteration, servers send updated values of the vertices present in the publish lists to the respective servers. After a scale-out/scale-in operation, we perform the publish–subscribe phase again to update the publish lists of servers.

First-Cut Migration

A first-cut approach is to perform migration of both static and dynamic data during the next available barrier synchronization interval. However, when we implemented this approach, we found that it added significant overheads by prolonging that iteration. As a result, we introduced two further optimizations as follows:

Static Data Migration

This technique is based on the observation that static data can be migrated in the background while computation is going on. Recall that static data consists of vertex IDs, their neighboring vertex IDs, and edge values to neighbors. Only dynamic data (vertex values and neighboring vertex values) need to wait to be migrated during a barrier synchronization interval (i.e., after such data are last updated). This reduces the overhead on that iteration.

Dynamic Data Migration

LFGraph has two barrier synchronization intervals. One interval is between the gather and scatter phases, and the other is after the scatter phase. That gives us two options for the transfer of dynamic data. We choose to perform dynamic data transfer and cluster reconfiguration in the barrier synchronization interval between the gather and scatter phases. This enables us to leverage the default scatter phase to migrate neighboring vertex values. The scatter phase simply considers the new set of servers in the cluster while distributing updated vertex values. This optimization further reduces the overhead on the iteration.

A scale-out/scale-in operation that starts in iteration i ends in iteration i + 2. Background static data migration occurs in iterations i and i + 1, while vertex value migration occurs after the gather phase of iteration i + 2. At that point, computation continues on the new set of servers. The performance impact due to background data migration is greater in iteration i than in iteration i + 1, that is, iteration times are longer in iteration i. The reason is that a majority of the migration happens in iteration i. In iteration i + 1 servers build their new subscription lists for the publish–subscribe phase.

To explain further, we will describe the steps involved in a scale-out as follows: (i) The joining server sends a Join message containing its IP address and port to the barrier server at the start of iteration i. (ii) The barrier server responds with a Cluster Info message assigning the joining server an ID and the contact information of the servers from which it should request its vertices. (iii) In addition, the barrier server sends an Add Host message to all servers, informing them about the new server in the cluster. (iv) The joining server requests its vertices with a Vertex Request message. (v) After receiving its vertices, it informs the barrier server with a Ready message that it can join the cluster. Concurrently, the servers start updating their subscription lists to reflect the modifications in the cluster servers. (vi) The barrier server sends a Reconfigure message to the servers in the synchronization interval after the gather phase of iteration i + 2. (vii) Upon receiving the Reconfigure message, joining servers request the vertex values with a Vertex Value Request message. In addition, all servers update their vertex-to-server mapping to reflect newly added servers. (viii) The scatter phase of iteration i + 2 executes with this new mapping. From then on, computation proceeds on the new set of servers.

Role of Barrier Server

In our repartitioning techniques, the barrier server accepts join and leave requests and determines an optimal partition assignment. We adopted this approach, instead of a fully decentralized reassignment, for two reasons: (i) fully decentralized reassignment may lead to complex race conditions, and (ii) the barrier server, once initialized, has the capability to obtain per-server iteration run times via the barrier synchronization messages and assigns new servers to alleviate the load on the busiest servers.

Experimental Setup

We performed our experiments with both our CVR and RVR techniques on virtual instances, each with 16 GB RAM and 8 VCPUs. We used a Twitter graph [130] containing 41.65 million vertices and 1.47 billion edges. (With larger graphs, we expect similar performance improvements.) We evaluated our techniques using five graph benchmarks: PageRank, single-source shortest paths (SSSP), connected components, k-means clustering, and multiple-source shortest paths (MSSP).

Scale-Out and Scale-In

Our first set of experiments measured the overhead experienced by the computation because of a scale-out operation. Figure 6.15 illustrates two experiments in which a scale-out from X servers to 2X servers (for X ∈ {5, 10, 15}) was performed, with the scale-out starting at iteration i = 1 and ending at iteration 3. The vertical axis plots the per-iteration run time. For comparison, we plot the per-iteration times for a run with X servers throughout, and a run with 2X servers throughout.

In Figure 6.15a–c, we can observe that (i) both CVR and RVR appear to perform similarly, and (ii) after the scale-out operation is completed, the performance of the scaled-out system converges to that of a cluster with 2X servers, demonstrating that our approaches converge to the desired throughput after scale-out.

Similarly, Figure 6.16 shows the plots for scale-in from 2X servers to X servers (for X ∈ {5, 10, 15}). Once again, the cluster converges to the performance of X servers.

Job Eviction Policies

The choice of victim job affects the completion time of lower priority research jobs by altering resources already allocated to them. Thus, job eviction policies need to be sensitive to current resource usage of individual research jobs. We discuss three resource-aware job eviction policies:

Most Resources (MR): This policy chooses as victim the research job that is currently using the most resources inside the cluster. In Hadoop YARN, resource usage would be in terms of the number of containers used by the job, while in other versions of MapReduce, it would be determined by the number of cluster slots.¹²

The MR policy, which is loosely akin to the worst-fit policy in OS segmentation, is motivated by the need to evict as few research jobs as possible; a large research job may contain sufficient resources to accommodate one large production job or multiple small production jobs. Thus, fewer research jobs are deferred, more of them complete earlier, and average research job completion time is minimized.

The downside of the MR policy is that when there is one large research job (as might be the case with heavy tailed distributions), it is always victimized whenever a production job arrives. This may lead to starvation and thus longer completion times for large research jobs.

Least Resources (LR): In order to prevent starving of large research jobs, this policy chooses as victim the research job that is currently using the least resources inside the cluster. The rationale here is that small research jobs that are preempted can always find resources if the cluster frees up even a little in the future. However, the LR policy can cause starvation for small research jobs if the cluster stays overloaded; for example, if a new production job arrives whenever one completes, LR will pick the same smallest jobs for eviction each time.

Probabilistically Weighted on Resources (PR): In order to address the starvation issues of LR and MR, our third policy selects a victim job using a probabilistic metric based on resource usage. In PR, the probability of choosing a job as a victim is directly proportional to the resources it currently holds. In effect, PR treats all tasks identically in choosing ones to evict, that is, if the task eviction policy were random, the chances of eviction for all tasks would be identical and independent of their jobs. The downside of PR is that it spreads out evictions across multiple jobs; in PR, unlike MR, one incoming production job may slow down multiple research jobs.

Task Eviction Policies

Once a victim job has been selected, the task eviction policy is applied within that job to select one task that will be preempted (i.e., suspended).

Our approach makes three assumptions, which are based on use case studies: (i) reduces are long enough that preemption of a task takes less time than the task itself; (ii) only reduce tasks are preempted; and (iii) reduces are stateless in between keys. For instance, in Facebook workloads the median reduce task takes 231 s [58], substantially longer than the time needed to preempt a task (see Section 6.1.5). There are two reasons why we focus on preemption only of reduces. First, the challenge of checkpointing reduce tasks subsumes that of checkpointing map tasks, since a map processes individual key-value pairs, while a reduce processes batches of them. Second, several use case studies have revealed that reduces are substantially longer than maps and thus have a bigger effect on the job tail. In the same Facebook trace already mentioned, the median map task time is only 19 s. While 27.1 map containers are freed per second, only 3 (out of 3100) reduce containers are freed per second. Thus, a small production job with 30 reduces would wait on average 10 s, and a large job with 3000 reduces would wait 1000 s. Finally, the traditional stateless reduce approach is used in many MapReduce programs; however, Natjam could be extended to support stateful reducers.

A MapReduce research job's completion time is determined by its last-finishing reduce task. A long tail, or even a single task that finishes late, will extend the research job's completion time. This concern implies that tasks with a shorter remaining time (for execution) must be evicted first. However, in multiprocessors, shortest-task-first scheduling is known to be optimal [53]. In our context, this means that the task with the longest remaining time must be evicted first. That motivated two contrasting task eviction policies, SRT and LRT:

Shortest Remaining Time ( SRT): In this policy, tasks that have the shortest remaining time are selected to be suspended. This policy aims to minimize the impact on the tail of a research job. Further, a task suspended by SRT will finish quickly once it has been resumed. Thus, SRT is loosely akin to the longest-task-first strategy in multiprocessor scheduling. Rather counterintuitively, SRT is provably optimal under certain conditions:

We note that the assumption – that tasks of the victim job will be resumed simultaneously – is reasonable in real-life scenarios in which production job submission times and sizes are unpredictable. Our experiments also validated this theorem.

Longest Remaining Time ( LRT): In this policy, the task with the longest remaining time is chosen to be suspended earlier. This policy is loosely akin to shortest-task-first scheduling in multiprocessors. Its main advantage over SRT is that it is less selfish and frees up more resources earlier. LRT might thus be useful in scenarios in which production job arrivals are bursty. Consider a victim job containing two tasks: one short and one with a long remaining time. SRT evicts the shorter task, freeing up resources for one production task. LRT evicts the longer task, but the shorter unevicted task will finish soon anyway, thus releasing resources for two production tasks, while incurring the overhead for only one task suspension. However, LRT can lengthen the tail of the research job, increasing its completion time.

Preemption in Hadoop YARN

Background: Hadoop YARN Architecture: In the Hadoop YARN architecture, a single cluster-wide Resource Manager (RM) performs resource management. It is assisted by one Node Manager (NM) per node (server). The RM receives periodic heartbeats from each NM containing status updates about resource usage and availability at that node.

The RM runs the Hadoop Capacity Scheduler. The Capacity Scheduler maintains multiple queues that contain jobs. An incoming job is submitted to one of these queues. An administrator can configure two capacities per queue: a minimum (guaranteed) capacity, and a maximum capacity. The scheduler varies the queue capacity between these two queues based on the jobs that have arrived at them.

The basic unit of resource allocation for a task is called a container. A container is effectively a resource slot that contains sufficient resources (primarily memory) to run one task: a map, a reduce, or a master task. An example master task is the Application Master (AM), which is allocated one container. One AM is assigned to each MapReduce job and performs job management functions.

An AM requests and receives, from the RM, container allocations for its tasks. The AM assigns a task to each container it receives and sends launch requests to the container's NM. It also performs speculative execution when needed.

An AM sends heartbeats to the RM. The AM also receives periodic heartbeats from its tasks. For efficiency, YARN piggybacks control traffic (e.g., container requests and task assignments) atop heartbeat messages.

Natjam Components: Natjam entails changes to the Hadoop Capacity Scheduler (at the RM) and the AM, while the NM stays unchanged. Specifically, Natjam adds the following new components to Hadoop YARN:

1. Preemptor: The preemptor is a part of the RM. We configured the Capacity Scheduler to contain two queues: one for production jobs and one for research jobs. The preemptor makes preemption decisions by using job eviction policies.
2. Releaser: This component is a part of each AM and is responsible for running the task eviction policies.

When we modify Hadoop, instead of adding new messages that will incur overhead, Natjam leverages and piggybacks atop YARN's existing heartbeats for efficiency. The trade-off is a small scheduling delay, but our experiments show that such delays are small.

We will detail those two components in Section 6.1.3. Now we show how preemption and checkpointing work.

Natjam Preemption Mechanism Example: We illustrate how Natjam's preemption works in YARN. Figure 6.17 depicts an example in which a research job 2 is initially executing in a full cluster, when a production Job 1 requires a single container.¹³ The steps in the figure are as follows:

1. Step 1: On AM1's heartbeat, Natjam asks the RM to allocate one container.
2. Steps 2 and 3: The cluster is full, so the RM applies the job eviction policies and selects Job 2 as victim.
3. Step 4: The Preemptor waits for AM2's next heartbeat, and in response to the heartbeat sends AM2 the number and type of containers to be released.
4. Step 5: The Releaser at AM2 uses the task eviction policy to select a victim task.
5. Step 6: When the victim task (still running) sends its next usual heartbeat to AM2, it is asked to suspend.
6. Step 7: The victim task suspends and saves a checkpoint.
7. Step 8: The victim task sends the checkpoint to AM2.
8. Step 9: The task indicates to NM-A that it has completed and it exits, freeing the container.

   With that, the Natjam-specific steps are done. For completeness, we list below the remaining steps, which are taken by YARN by default to give AM1 the new container.

9. Step 10:NM-A's heartbeat sends the container to RM.
10. Step 11:AM1's next RM heartbeat gets the container.
11. Step 12: AM1 sends NM-A the task request.
12. Step 13: NM-A launches the task on the container.

Checkpoint Saved and Used by Natjam: When Natjam suspends a research job's reduce task, an on-demand checkpoint is saved automatically. It contains the following items: (i) an ordered list of past suspended container IDs, one for each attempt, that is, each time this task was suspended in the past; (ii) a key counter, that is, the number of keys that have been processed so far; (iii) reduce input paths, that is, local file paths; and (iv) the hostname associated with the last suspended attempt, which is useful for preferably resuming the research task on the same server. Natjam also leverages intermediate task data already available via Hadoop [136], including (v) reduce inputs, which are stored at a local host, and (vi) reduce outputs, which are stored on HDFS.

Task Suspend: We modified YARN so that the reduce task keeps track of two pieces of state: paths to files in the local file system that hold reduce input and the key counter, that is, the number of keys that have been processed by the reduce function so far. When a reduce task receives a suspend request from its AM, the task checks whether it is in the middle of processing a particular key, and finishes that key. Second, it writes the input file paths to a local log file. Third, Hadoop maintains a partial output file per reduce attempt, in the HDFS distributed file system. It holds the output so far generated from the current attempt. The partial output file is given a name that includes the container ID. When a task suspends, this partial output file is closed. Finally, the reduce compiles its checkpoint and sends the result to its AM, and the reduce task exits.

Task Resume: On a resume, the task's AM sends the saved checkpoint state as launch parameters to the chosen NM. The Preemptor is in charge of scheduling the resuming reduce on a node. The Preemptor prefers the old node on which the last attempt ran (available from the hostname field in the checkpoint). If the resumed task is assigned to its old node, the reduce input can be read without network overhead, that is, from local disk. If it is resumed on a different node, the reduce input is assembled from map task outputs, much like a new task.

Next, the reduce task creates a new partial output file in HDFS. It skips over the input keys that the checkpoint's key counter field indicates have already been processed. It then starts execution as a normal reduce task.

Commit after Resume: When a previously suspended reduce task finishes, it needs to assemble its partial output. It starts that by finding, in HDFS, all its past partial output files; it does so by using the ordered list of past suspended container IDs from its checkpoint. It then accumulates their data into output HDFS files that are named in that order. This order is critical so that the output is indistinguishable from that of a reduce task that was never suspended.

Implementation Issues

This section first explains how we modify the AM state machines in Hadoop YARN, and then describes the Preemptor and Releaser. As mentioned earlier, we leverage existing Hadoop mechanisms such as heartbeats.

Application Master's State Machines: For job and task management, Hadoop YARN's AM maintains separate state machines per job, per task, and per task attempt. Natjam does not change the job state machine; we enabled this state machine only to handle the checkpoint. Thus, both the suspend and the resume occur during the Running state in this state machine.

We modify the task state machine very little. When the AM learns that a task attempt has been suspended (from step 8 in Figure 6.17), the task state machine goes ahead and creates a new task attempt to resume the task. However, this does not mean that the task is scheduled immediately; the transitions of the task attempt state machine determine whether it does.

The task attempt state machine is used by YARN to assign the container, set up execution parameters, monitor progress, and commit output. Natjam adds two states to the task attempt state machine, as shown in Figure 6.18: Suspend-Pending and Suspended. The task attempt has a state of Suspend-Pending when it wishes to suspend a task but has not received suspension confirmation from the local task (steps 6b–7 from Figure 6.17). The state becomes Suspended when the saved checkpoint is received (step 8), and this is a terminal state for that task attempt.

The new transitions for suspension in Figure 6.18 are as follows:

• S1: AM asks the task to suspend and requests a checkpoint.
• S2: AM receives the task checkpoint and saves it in the task attempt state machine.

A resuming reduce task starts from the New state in the task attempt state machine. However, we modify some transitions to distinguish a resuming task from a new (nonresuming) task attempt as follows:

• R1: Just as for any reduce task attempt, every heartbeat from the AM to RM requests a container for the resuming reduce. If the RM cannot satisfy the request, it ignores it (since the next heartbeat will resend the request anyway). Suppose a container frees up (e.g., as production jobs complete), so that the RM can now schedule a research task. In doing so, the RM prefers to respond to a resuming reduce's request, rather than one from a non-resuming research reduce. The AM to RM requests also carry the hostname field from the task checkpoint; the RM uses it to declare a preference for allocating that container at that hostname.
• R2: Once the AM has received a container from the RM, it launches a task attempt on the allocated container. For resuming reduces, the AM also sends the saved checkpoint to the container.
• C1: On commit, the AM accumulates partial output files into the final task output in HDFS (see Section 6.1.3).

Preemptor: Recall that Natjam sets up the RM's Capacity Scheduler with two queues: one for production jobs and one for research jobs. The Preemptor is implemented as a thread within the Capacity Scheduler. In order to reclaim resources from the research queue for use by the production queue, the Preemptor periodically runs a reclaim algorithm, with sleeps of 1 s between runs. A run may generate reclaim requests, each of which is sent to some research job's AM to reclaim a container (which is step 4 in Figure 6.17). In a sense, a reclaim request is a statement of a production job's intention to acquire a container.

We keep track of a per-production job reclaim list. When the RM sends a reclaim request on behalf of a job, an entry is added to the job's reclaim list. When a container is allocated to that job, that reclaim list entry is removed. The reclaim list is needed to prevent the Preemptor from generating duplicate reclaim requests, which might occur because our reliance on heartbeats entails a delay between a container suspension and its subsequent allocation to a new task. Thus, we generate a reclaim request whenever (i) the cluster is full, and (ii) the number of pending container requests from a job is greater than the number of requests in its reclaim list.

In extreme cases, the Preemptor may need to kill a container, for example, if the AM has remained unresponsive for too long. Our threshold to kill a container is reached when a reclaim request has remained in the reclaim list for longer than a killing timeout (12 s). A kill request is sent directly to the NM to kill the container. This bypasses the AM, ensuring that the container will indeed be killed. When a kill request is sent, the reclaim request is added to an expired list. It remains there for an additional time interval (2 s), after which it is assumed the container is dead, and the reclaim request is thus removed from the expired list. With those timeouts, we have never observed killings of any tasks in any of our cluster runs.

Releaser: The Releaser runs at each job's AM and decides which tasks to suspend. Since the task eviction policies discussed in Section 6.1.2 (i.e., SRT and LRT) use the time remaining at the task, the Releaser needs to estimate it. We use Hadoop's default exponentially smoothed task runtime estimator, which relies on the task's observed progress [69]. However, calculating this estimate on demand can be expensive due to the large numbers of tasks. Thus, we have the AM only periodically estimate the progress of all tasks in the job (once a second), and use the latest complete set of estimates for task selection. While the estimates might be stale, our experiments show that this approach works well in practice.

Interaction with Speculative Execution: Our discussion so far has ignored speculative execution, which Hadoop uses to replicate straggler task attempts. Natjam does not change speculative execution and works orthogonally, that is, speculative task attempts are candidates for eviction. When all attempts of a task are evicted, the progress rate calculation of the task is not skewed, because speculative execution tracks the progress of task attempts rather than the tasks themselves. While this interaction could be optimized further, it works well in practice. Natjam can be further optimized to support user-defined (i.e., per job) task eviction policies that would prioritize the eviction of speculative tasks, but discussion of that is beyond the scope of this chapter.

Eviction Policies

First, for job eviction, we explored two deadline-based policies inspired by the classical real-time literature [78,81]: Maximum Deadline First (MDF) and Maximum Laxity First (MLF). MDF chooses as victim the running job that has the longest deadline. On the other hand, MLF evicts the job with the highest laxity, where laxity = the deadline minus the job's projected completion time. For MLF, we extrapolate Hadoop's reported job progress rate to calculate a job's projected completion time.

While MDF is a static scheduling policy that accounts only for deadlines, MLF is a dynamic policy that also accounts for a job's resource needs. If a job has an unsatisfactory progress rate, MLF may give it more resources closer to its deadline. It may do so by evicting small jobs with long deadlines. In essence, while MLF may run some long-deadline, high-resource jobs, MDF might starve all long-deadline jobs equally. Further, MLF is fair in that it allows many jobs with similar laxities to make simultaneous progress. However, this fairness can be a shortcoming in scenarios with many short deadlines; MLF results in many deadline misses, while MDF would meet at least some deadlines. Section 6.1.6 describes our experimental evaluation of this issue.

Second, our task eviction policies remain the same as before (SRT, LRT) because the deadline is for the job, not for individual tasks.

In addition to the job and task eviction policies, we need to have a job selection policy. When resources free up, this policy selects a job from among the suspended ones and gives it containers. Possible job selection policies are earliest deadline first (EDF) and least laxity first (LLF). In fact, we implemented these but observed thrashing-like scheduling behavior if the job eviction policy was inconsistent with the job selection policy. For instance, if we used MDF job eviction and LLF job selection, a job selected for eviction by MDF would soon after be selected for resumption by LLF, and thus enter a suspend-resume loop. We concluded that the job selection policy needed to be dictated by the job eviction policy, that is, MDF job eviction implies EDF job selection, while MLF implies LLF job selection.

Implementation

The main changes that differentiate Natjam-R from Natjam are in the RM. In Natjam-R, the RM keeps one Capacity Scheduler queue sorted by decreasing priority. A priority is inversely proportional to the deadline for MDF, and to laxity for MLF. The Preemptor periodically (once a second) examines the queue and selects the first job (say J_i) that still has tasks waiting to be scheduled. Then it considers job eviction candidates from the queue, starting with the lowest priority (i.e., later deadlines or larger laxities) up to J_i's priority. If it encounters a job that still has allocated resources, that job is picked as the victim; otherwise, no further action is taken. To evict the job, the Releaser from Natjam uses the task eviction policy to free a container. Checkpointing, suspend, and resume work in Natjam-R as described earlier for Natjam (see Section 6.1.3).

Experimental Plan

We present two sets of experiments, increasing in complexity and scale. This section presents microbenchmarking results for a small Natjam cluster. Section 6.1.6 evaluates Natjam-R. While we present only selected experimental results in this chapter, we refer the reader to Ref. [131] for extensive information on experiments and evaluation of our system.

Microbenchmark Setup

We first evaluated the core Natjam system that supports a dual-priority workload, that is, research and production jobs. We addressed the following questions: (i) How beneficial is Natjam relative to existing techniques? (ii) What is the overhead of the Natjam suspend mechanism? (iii) What are the best job eviction and task eviction policies?

We used a small-scale test bed and a representative workload because this first experimental stage involved exploration of different parameter settings and study of many fine-grained aspects of system performance. A small test bed gave us flexibility.

Our test cluster had seven servers running on a 1 GigE network. Each server had two quad-core processors and 16 GB of RAM, of which 8 GB were configured to run 1 GB-sized Hadoop containers. (Thus, 48 containers were available in the cluster.) One server acted as the Resource Manager, while the other six were workers. Each entity (AM, map task, and reduce task) used one container.

In our experiments, we injected a mix of research and production jobs, as shown in Table 6.4. To reflect job size variation, the job sizes ranged from XL (filling the entire cluster) to S (filling a fourth of the cluster). To mimic use case studies [58], each job had a small map execution time, and was dominated by the reduce execution time. To model variance in task running times, we selected reduce task lengths uniformly from the interval (0.5, 1.0], where 1.0 is the normalized largest reduce task. To emulate computations, we used SWIM [137] to create random keys and values, with thread sleeps called between keys. Shuffle and HDFS traffic were incurred as usual.

The primary metric was job completion time. Each of our data points shows an average and standard deviation over five runs. Unless otherwise noted, Natjam used MR job eviction and SRT task eviction policies.

Natjam versus Existing Techniques

Figure 6.19 compares Natjam to several alternatives versus an ideal setting versus two existing mechanisms in the Hadoop Capacity scheduler, and versus pessimistic killing of tasks (instead of saving the cheap checkpoint). The ideal setting measures each job's completion time when it is executed on an otherwise empty cluster; thus, it ignores resource sharing and context switch overheads. For the second setting, we chose the Hadoop Capacity Scheduler because it represents approaches that we might take with two physically separate clusters sharing the same scheduler. Finally, killing of tasks is akin to approaches such as those described in Ref. [59] and for the Hadoop Fair Scheduler [84].

In this experiment, a Research-XL job was submitted initially to occupy the entire cluster. Then, 50 s later, a Production-S job was submitted. Figure 6.19 shows that killing of tasks (the fourth pair of bars) finished production jobs fast, but prolonged research jobs by 23% compared to the ideal case (the first pair of bars). Thus, saving the overhead of checkpoints is not worth the repeated work due to task restarts.

We next examined two popular Hadoop Capacity Scheduler approaches called Hard cap and Soft cap. Recall that the Capacity Scheduler allows the administrator to set a maximum cap on the capacity allocated to each of the two queues (research and production). In Hard cap, that cap is used as a hard limit for each queue. In the Soft cap approach, each queue is allowed to expand to the full cluster if there are unused resources, but it cannot scale down without waiting for its scheduled tasks to finish (e.g., if the production queue needs resources from the research queue). We configured these two approaches with the research queue set to 75% capacity (36 containers) and production queue to 25% capacity (12 containers), as these settings performed well.

Figure 6.19 shows that in Hard cap (the second pair of bars), the research job took 52% longer than ideal, while the production job was unaffected. Under Soft cap (the third pair of bars), the production job could obtain containers only when the research job freed them; this resulted in an 85% increase in production job completion time, while the research job was unaffected.

The last pair of bars shows that when Natjam was used, the production job's completion time was 7% worse (5.4 s longer) than ideal and 77% better than the result for the Hadoop Capacity Scheduler's Soft cap. The research job's completion time was only 2% worse (4.7 s longer) than ideal, 20% better than that of killing, and 49% better than that of Hadoop Hard cap. One of the reasons the research job was close to ideal is that it was able to make progress in parallel with the production job. There are other internal reasons for the performance benefit, which we explore next.

Suspend Overhead

We measured Natjam's suspend overhead on a fully loaded cluster. We observed that it took an average of 1.35 s to suspend a task and 3.88 s to resume a task. Standard deviations were low. In comparison, default Hadoop took an average of 2.63 s to schedule a task on an empty cluster. From this it might appear that Natjam incurs a higher total overhead of 5.23 s per task suspend-resume. However, in practice the effective overhead is lower; for instance, Figure 6.19 showed only a 4.7 s increase in research job completion time. The reason is that task suspends typically occur in parallel, and in some cases, task resumes do too. Thus, the time overheads are parallelized rather than aggregated.

Task Eviction Policies

We now compare the two task eviction policies (SRT and LRT) from Section 6.1.2 against each other, and against a random eviction strategy that we also implemented. We performed two sets of experiments: one with Production-S and another with Production-L. The production job was injected 50 s after a Research-XL job.

Table 6.5 tabulates the results. In all cases the production job incurred overhead similar to that for an empty cluster. Thus, we discuss only research job completion time (last column). As shown in the top half of the table, a random task eviction strategy resulted in a 45 s increase in completion time compared to the ideal; we observed that a fourth of the tasks were suspended, leading to a long job tail. Evicting the LRT incurred a higher increase of 55 s because LRT prolongs the tail. Evicting the SRT emerged as the best policy and was only 4.7 s worse than the ideal because it respects the job tail.

In the lower half of Table 6.5, it can be seen that a larger production job caused more suspensions. The research job completion times for the random and LRT eviction policies are similar to those in the top half because the job's tail was already long for the small production job, and was not much longer for the larger job. SRT is worse than it was for a small production job, yet it outperformed the other two eviction strategies.

We conclude that SRT is the best task eviction policy, especially when production jobs are smaller than research jobs. We believe this is a significant use case since research jobs run longer and process more data, while production jobs are typically small due to the need for faster results.

Job Eviction Policies

We next compare the three job eviction policies discussed in Section 6.1.2. Based on the previous results, we always used SRT task eviction.¹⁴ We initially submitted two research jobs and followed 50 s later with a small production job. We examined two settings: one in which the initial research jobs were comparable in size and another in which they were different. We observed that the production job completion time was close to ideal; hence Table 6.6 shows only research job completion times.

The top half of Table 6.6 shows that when research job sizes were comparable, probabilistic weighing of job evictions by resources (PR) and eviction of the job with the MR performed comparably: research job completion times for the two policies were within 2 s (0.5%) of each other. This is desirable due to the matching job sizes. On the other hand, eviction of the job with the LR performed the worst, because it caused starvation in one of the jobs. Once tasks start getting evicted from a research job (which may at first have been picked randomly by LR if all jobs had the same resource usage), subsequently LR will always pick the same job (until it is fully suspended).

That behavior of LR is even more pronounced on small research jobs in a heterogeneous mix, as can be seen in the bottom half of Table 6.6. The Research-S job is picked as a victim by PR less often than by LR, and thus PR outperforms LR. PR penalizes the Research-L job slightly more than LR does, since PR evicts more tasks from a larger job. Even so, PR and MR are within 10 s (5%) of each other; any differences are due to the variable task lengths, and the effectiveness of the SRT task eviction policy. We observed that MR evicted no tasks at all from the Research-S job.

We conclude that when the best task eviction policy (SRT) is used, the PR and MR job eviction policies are preferable to LR, and MR is especially good under heterogeneous mixes of research job sizes.

MDF versus MLF

We injected three identical jobs, Job 1 to Job 3, each with 8 maps and 50 reduces. (Each job took 87 s on an empty cluster.) They were submitted in numerical order starting at t = 0 s and 5 s apart, thus overloading the cluster. Since MDF and MLF will both meet long deadlines, we chose shorter deadlines. To force preemption, the deadlines of job 1, job 2, and job 3 were set 10 s apart: 200 s, 190 s, and 180 s, respectively.

Figure 6.20 depicts the progress rate for the MDF cluster and the MLF cluster. Our first observation is that while MDF allowed the short-deadline jobs to run earlier and thus satisfy all deadlines, MLF missed all deadlines (see Figure 6.20b). In the reduce phase for MLF, after a while jobs proceeded in lockstep in the reduce phase, because when a lower laxity job (e.g., job 3) has run for a while in lieu of a higher laxity job (e.g., job 1), their laxities become comparable. Thereafter, the two jobs take turns preempting each other. Breaking ties, for example, by using a deadline, does not eliminate this behavior. In a sense, MLF tries to be fair to all jobs by allowing them all to make progress simultaneously, but this fairness is in fact a drawback.

MLF also takes longer to finish all jobs, that is, 239 s compared to MDF's 175 s. MLF's lockstep behavior incurs a high context switch overhead. We conclude that MDF is preferable to MLF, especially under short deadlines.

Varying the Deadline

We submitted a job (job 1) just as described previously, and 5 s later submitted an identical job (job 2) whose deadline was 1 s earlier than job 1's. We measured job 1's clean compute time as the time to run the job in an empty cluster. Then, we set its deadline = submission time + (clean compute time × (1 + ɛ)). Figure 6.21 shows the effect of ɛ on a metric called margin. We define a job's margin =(deadline) minus (job completion time). A negative margin implies a deadline miss. We observe that an ε as low as 0.8 still meets both deadlines, while an ɛ as low as 0.2 meets at least the shorter deadline. This means that given one critical job with a very short deadline, Natjam-R can satisfy it if it has at least 20% more time than the job's clean compute time. This percentage is thus an estimate of Natjam-R's overhead. We also performed experiments that varied the second job's size as a fraction of the first job from 0.4 to 2.0, but we saw little effect on margin.

Trace-Driven Experiments

We used the Yahoo! Hadoop traces to evaluate Natjam-R's deadline satisfaction. We used only the production cluster trace, scaled so as to overload the target cluster. Since the original system did not support deadline scheduling, no deadlines were available from the traces. Thus, we chose ε randomly for each job from the interval [0, 2.0], and used it to set the job's deadline forward from its submission time (as described earlier). A given job's deadline was selected to be the same in all runs.

Figure 6.22 compares Natjam-R against Hadoop Soft cap. It shows the CDF of the difference between the margins of the two approaches; a negative difference implies that Natjam-R is better. Natjam-R's margin is better than Soft cap's for 69% of jobs. The largest improvement in margin was 366 s. The plot is biased by one outlier job that took 1000 s longer in Natjam-R; the next greatest outlier is only –287 s. The first outlier job suffered in Natjam-R because the four jobs submitted just before it and one job right after had much shorter deadlines. Yet the conclusion is positive: among the 400 jobs with variable deadlines, there was only one such outlier. We conclude that Natjam-R satisfies deadlines well under a realistic workload.