The performance of modern Big Data systems used for data processing is often bottlenecked by data movement across various components.Offloading part of the processing closer to the storage and network to reduce this bottleneck is a compelling idea, and in today's hardware landscape, there are many different types of Smart Storage, Smart NIC, or Smart Switch devices one could choose from.One challenge, however, is that it is often unclear at design time what improvements of the end workload one can achieve with a given hardware.Co-design is typically mentioned in related work as the solution, but it is far from obvious what this entails in practice: what information about the hardware, software system, and workload is taken into account for decision-making is often implicit in related work.In this work, we propose a model-driven methodology for rightsizing offloading to benefit an end workload.Our methodology determines the target processing rate an offload device should have in a specific data processing system, without over-fitting.This helps designers pick the right hardware for offloading.Our methodology relies on modeling the system as a network of queues, based on different levels of information about the system, which allows determining the general usefulness of offload and the specific benefits to a workload of interest.We demonstrate how our methodology avoids under-or over-provisioning offload devices in a case study.

Storage disaggregation underlies today's cloud and is naturally complemented by pushing down some computation to storage, thus mitigating the potential network bottleneck between the storage and compute tiers. We show how ML training benefits from storage pushdowns by focusing on transfer learning (TL), the widespread technique that democratizes ML by reusing existing knowledge on related tasks. We propose HAPI, a new TL processing system centered around two complementary techniques that address challenges introduced by disaggregation. First, applications must carefully balance execution across tiers for performance. HAPI judiciously splits the TL computation during the feature extraction phase yielding pushdowns that not only improve network time but also improve total TL training time by overlapping the execution of consecutive training iterations across tiers. Second, operators want resource efficiency from the storage-side computational resources. HAPI employs storage-side batch size adaptation allowing increased storage-side pushdown concurrency without affecting training accuracy. HAPI yields up to 2.5× training speed-up while choosing in 86.8% of cases the best performing split point or one that is at most 5% off from the best.

State management has long been a challenge for serverless applications. Owing to their failure resilience and consistency guarantees, distributed shared logs have been recently proposed as a promising storage substrate enabling stateful serverless applications. We show that, unfortunately, state-of-the-art sacrifices compute tier scalability for log access performance, a particularly undesirable exchange for the dynamic serverless environment. The culprit is the log indexing architecture, namely relying on complete local indexes colocated with serverless functions. This design prevents efficient scaling and even risks out-of-memory errors.

Stateful serverless applications need to persist their state and data. The existing approach is to store the data in general purpose storage systems. However, these approaches are not designed to meet the demands of serverless applications in terms of consistency, fault tolerance and performance.

Hardware-induced performance variability has long been an undesirable fact of life in the storage stack. SSDs have not managed to break the trend. Despite continuous evolution in their internal design, SSD performance variability has remained a concern. This unfortunate trend is also bound to continue for SSDs for the foreseeable future due to increasingly complex controller design and responsibilities. Nevertheless, applications demand high, predictable and stable performance. It is therefore important to measure and understand the performance implications of this hardware-induced variability at the application layer. It is equally vital to assess to what extent the mechanisms available at the software level are able to alleviate or mask the variability. In this paper we uncover and analyze three novel and surprising performance anomalies induced by SSDs. We focus on reads. At the application layer, each anomaly leads to significant read throughput slowdown. The first anomaly, intrinsic slowdown, slows down reads for a variable amount of time when reading from new file system extents. Second, temporal slowdown slows down reads periodically, even in the absence of any writes. Third, in permanent slowdown, reads from some files eventually become consistently slow and never recover. Individually, each of the slowdowns can cause a read throughput loss of 10%-15%, but when they occur concurrently the cumulative throughput loss can reach 30%. We further analyze to what extent available software mechanisms can mask these performance anomalies. We find that only two of the three slowdowns can be masked via increased I/O request parallelism.

Migrating data into memory can significantly accelerate big-data applications by hiding low disk throughput. While prior work has mostly targeted caching frequently used data, the techniques employed do not benefit jobs that read cold data. For these jobs, the file system has to pro-actively migrate the inputs into memory. Successfully migrating cold inputs can result in a large speedup for many jobs, especially those that spend a significant part of their execution reading inputs. In this paper, we use data from the Google cluster trace to make the case that the conditions in production workloads are favorable for migration. We then design and implement DYRS, a framework for migrating cold data in big-data file systems. DYRS can adapt to match the available bandwidth on storage nodes, ensuring all nodes are fully utilized throughout the migration. In addition to balancing the load, DYRS optimizes the placement of each migration to maximize the number of successful migrations and eliminate stragglers at the end of a job. We evaluate DYRS using several Hive queries, a trace-based workload from Facebook, and the Sort application. Our results show that DYRS successfully adapts to bandwidth heterogeneity and effectively migrates data. DYRS accelerates Hive queries by up to 48%, and by 36% on average. Jobs in a trace-based workload experience a speedup of 33% on average. The mapper tasks in this workload have an even greater speedup of 46%. DYRS accelerates sort jobs by up to 20%.

The vast majority of data center schedulers use task runtime estimates to improve the quality of their scheduling decisions. Knowledge about runtimes allows the schedulers, among other things, to achieve better load balance and to avoid head-of-line blocking. Obtaining accurate runtime estimates is, however, far from trivial, and erroneous estimates lead to sub-optimal scheduling decisions. Techniques to mitigate the effect of inaccurate estimates have shown some success, but the fundamental problem remains.

This paper investigates whether migrating cold data can yield significant speedup for big data jobs that run on modern big data file systems. Our work is motivated by two observations. First, improving the input stage of a job can provide significant speedup because many jobs spend a large part of their execution reading inputs. The second observation is that the inputs for many jobs are cold. Common techniques that aim to keep hot data in memory do not benefit these jobs. We analyze the Google production cluster trace data and find that the key ingredients for effectively migrating cold data do exist in such production environments. Encouraged by our findings, we design and implement Ignem, a framework for migrating cold data in big data file systems. We evaluate Ignem in a series of experiments and show that it provides significant speedup for both small and large jobs. Specifically, Hive queries are accelerated by up to 34%; the mean job duration in a trace-driven workload is reduced by 12% and the task duration by nearly 40%; other standalone jobs such as sort and wordcount also improve similarly by up to 30%.

Log-Structured Merge Key-Value stores (LSM KVs) are designed to offer good write performance, by capturing client writes in memory, and only later flushing them to storage. Writes are later compacted into a tree-like data structure on disk to improve read performance and to reduce storage space use. It has been widely documented that compactions severely hamper throughput. Various optimizations have successfully dealt with this problem. These techniques include, among others, rate-limiting flushes and compactions, selecting among compactions for maximum effect, and limiting compactions to the highest level by so-called fragmented LSMs. In this article, we focus on latencies rather than throughput. We first document the fact that LSM KVs exhibit high tail latencies. The techniques that have been proposed for optimizing throughput do not address this issue, and, in fact, in some cases, exacerbate it. The root cause of these high tail latencies is interference between client writes, flushes, and compactions. Another major cause for tail latency is the heterogeneous nature of the workloads in terms of operation mix and item sizes whereby a few more computationally heavy requests slow down the vast majority of smaller requests. We introduce the notion of an Input/Output (I/O) bandwidth scheduler for an LSM-based KV store to reduce tail latency caused by interference of flushing and compactions and by workload heterogeneity. We explore three techniques as part of this I/O scheduler: (1) opportunistically allocating more bandwidth to internal operations during periods of low load, (2) prioritizing flushes and compactions at the lower levels of the tree, and (3) separating client requests by size and by data access path. SILK+ is a new open-source LSM KV that incorporates this notion of an I/O scheduler.

We present Eagle, a new hybrid data center scheduler for data-parallel programs. Eagle dynamically divides the nodes of the data center in partitions for the execution of long and short jobs, thereby avoiding head-of-line blocking. Furthermore, it provides job awareness and avoids stragglers by a new technique, called Sticky Batch Probing (SBP).

The performance of HDFS is critical to big data software stacks and has been at the forefront of recent efforts from the industry and the open source community. A key problem is the lack of flexibility in how data replication is performed. To address this problem, this paper presents Pfimbi, the first alternative to HDFS that supports both synchronous and flow-controlled asynchronous data replication. Pfimbi has numerous benefits: It accelerates jobs, exploits under-utilized storage I/O bandwidth, and supports hierarchical storage I/O bandwidth allocation policies. We demonstrate that for a job trace derived from a Facebook workload, Pfimbi improves the average job runtime by 18% and by up to 46% in the best case. We also demonstrate that flow control is crucial to fully exploiting the benefits of asynchronous replication; removing Pfimbi's flow control mechanisms resulted in a 2.7× increase in job runtime.

Data replication, the main failure resilience strategy used for big data analytics jobs, can be unnecessarily inefficient. It can cause serious performance degradation when applied to intermediate job outputs in multi-job computations. For instance, for I/O-intensive big data jobs, data replication is especially expensive because very large datasets need to be replicated. Reducing the number of replicas is not a satisfactory solution as it only aggravates a fundamental limitation of data replication: its failure resilience guarantees are limited by the number of available replicas. When all replicas of some piece of intermediate job output are lost, cascading job recomputations may be required for recovery. In this paper we show how job recomputation can be made a first-order failure resilience strategy for big data analytics. The need for data replication can thus be significantly reduced. We present RCMP, a system that performs efficient job recomputation. RCMP can persist task outputs across jobs and leverage them to minimize the work performed during job recomputations. More importantly, RCMP addresses two important challenges that appear during job recomputations. The first is efficiently utilizing the available compute node parallelism. The second is dealing with hot-spots. RCMP handles both by switching to a finer-grained task scheduling granularity for recomputations. Our experiments show that RCMP's benefits hold across two different clusters, for job inputs as small as 40GB or as large as 1.2TB. Compared to RCMP, data replication is 30%-100% worse during failure-free periods. More importantly, by efficiently performing recomputations, RCMP is comparable or better even under single and double data loss events.

Hadoop has become a critical component in today's cloud environment. Ensuring good performance for Hadoop is paramount for the wide-range of applications built on top of it. In this paper we analyze Hadoop's behavior under failures involving compute nodes. We find that even a single failure can result in inflated, variable and unpredictable job running times, all undesirable properties in a distributed system. We systematically track the causes underlying this distressing behavior. First, we find that Hadoop makes unrealistic assumptions about task progress rates. These assumptions can be easily invalidated by the cloud environment and, more surprisingly, by Hadoop's own design decisions. The result are significant inefficiencies in Hadoop's speculative execution algorithm. Second, failures are re-discovered individually by each task at the cost of great degradation in job running time. The reason is that Hadoop focuses on extreme scalability and thus trades off possible improvements resulting from sharing failure information between tasks. Third, Hadoop does not consider the causes of connection failures between its tasks. We show that the resulting overloading of connection failure semantics unnecessarily causes an otherwise localized failure to propagate to healthy tasks. We also discuss the implications of our findings and draw attention to new ways of improving Hadoop-like frameworks.

This paper proposes a purely passive method for inferring a congestion map of a network. The congestion map is computed using the congestion markings carried in existing traffic, and is continuously updated as traffic is received. Consequently, congestion changes can be tracked in a real-time fashion with zero traffic overhead. Unlike active congestion reporting methods, our novel passive method is more robust during periods of congestion because there are no congestion report messages that could be lost and existing congestion is never aggravated. Our solution has several applications ranging from informing IP fast re-route algorithms and traffic engineering schemes to assisting in inter-domain path selection.

This paper presents the CONTRACT framework to address a fundamental deficiency of the IP network control plane, namely the lack of coordination between an IGP and other control functions involved in achieving a high level objective. For example, an IGP's default automatic reaction to a network failure may result in an SLA violation, even if the IGP link weights have been carefully chosen. This is because an IGP blindly routes traffic along the shortest paths based on link weights, and it is completely oblivious to the interactions between SLA compliance, load balancing and traffic policing objectives in a network. The CONTRACT framework makes it possible to coordinate these objectives. Under this framework, routers continue to operate autonomously, but they also coordinate their actions with a centralized network controller, which evaluates the impact of routing changes, decides whether the changes are SLA compliant, and performs load rebalancing and/or packet filter reconfiguration as necessary. The key contribution of CONTRACT is a set of coordination algorithms. We show that CONTRACT can effectively coordinate the actions of routing, load balancing and traffic policing to improve a network's SLA compliance.