To cope with the increasing number of Internet users as well as Internet-of-Things (IoT) and edge devices, large organizations leverage tens to hundreds of datacenters and edge sites. Collecting data related to end-user sessions, monitoring logs, and performance counters, and then analyzing and personalizing this data can significantly improve overall user experience. The key challenge, however, is posed by the wide-area network (WAN) that connects the distributed sites. Unlike datacenter networks, WAN bandwidth is significantly lower and latency is higher; both are highly variable too. The overarching goal of the project was to redesign big data analytics stacks to support timely and efficient analysis, with a specific focus on analysis over geo-distributed sites. As part of this project, we built a suite of solutions to rethink storage, query planning, execution engine, and networking for geo-distributed analytics (GDA).

To enable general-purpose, interactive GDA across distributed sites, we designed a federated execution engine, Sol, that is API-compatible with Apache Spark. Meaning, any existing jobs can now run across multiple datacenters in an interactive fashion. Sol relies on two key changes from traditional datacenter-based execution engines. First, it pushes task instead of waiting for workers to be ready, which saves latency. Second, it decouples the provisioning of resources for communication and computation to achieve high resource utilization. As a result, it speeds up SQL and machine learning jobs by 4.9X and 16.4X compared to Apache Spark in resource-constrained networks. Moreover, even in datacenter environments, Sol outperforms Spark by 1.3X to 3.9X. 

To manage the large data transfers between multiple datacenters for large, non-interactive GDA jobs, we designed the best-to-date communication scheduling-routing solution (NOCS). Specifically, we provided a tight 2-approximation algorithm when all release times and demands are polynomially sized, and a (2+ϵ)-approximation when the release times and demands can be super-polynomial. We have also developed variations of solutions that are computationally faster but may not always have well-analyzed approximation ratios. Evaluations of these algorithms show that the performance improvement in practice is even better than that predicted in theory.

We also made important progress in memory and storage management (EC-Cache) as well as dynamic query planning (QOOP) as part of this project. Specifically, EC-Cache is a low-latency erasure-coded (in-memory) storage that can improve efficiency and load balancing without sacrificing fault tolerance and availability for both datacenter and GDA jobs. We show that a small bandwidth overhead of at most 10% can provide more than 50% reduction in the median and tail latencies. EC-Cache's latency reductions increase as objects grow larger: for example, 1.33X for 1 MB objects and 5.5X for 100 MB objects. QOOP focuses on dynamic query (re)planning in changing conditions, such as WAN bandwidth variability. When resource availability changes, QOOP evaluates the progress of a query and picks a better query plan to switch to (if one exists). Dynamically reacting to such changing conditions can offer a median performance improvement of 1.47X compared to state-of-the-art alternatives.

Finally, we have explored two specific use case of GDA applications in CellScope and System H. CellScope enables geo-distributed machine learning for radio access network (RAN) performance diagnostics by trading of timeliness and accuracy. Our implementation on Apache Spark and evaluation shows that CellScope is able to achieve accuracy improvements up to 4.4X without incurring the latency overhead associated with the existing approaches. System H focuses on hybrid multi-cloud analytics (HMCA), where an enterprise wants to run analytics across public and private clouds owned by multiple entities. By taking WAN topology into account, we show that it can outperform the state-of- the-art by up to 3.43X for the smallest and up to 32.04X for the largest deployment on industry-standard TPC benchmarks.

All software developed as part of this project are based on established open-source systems such as Apache Spark and Apache YARN, and we have and continue to open-source our works at https://github.com/symbioticlab. Research papers summarizing our works have been published or are under submission in top venues in networking and systems including OSDI, NSDI, MobiCom, and SPAA. Some of the works have been incorporated into course contents in graduate- and undergraduate-level systems and networking courses at the University of Michigan. Last but not the least, two PhD students at the University of Michigan have worked on different pieces of our contributions, and this grant has helped in partly supporting their education and training.