Hybrid Cloud and HPC Approach to High-Performance Dataframes

The work presented in this paper is an extension of Cylon. Cylon is a high-performance distributed-memory framework parallelized using the Bulk Synchronous Parallel (BSP) pattern [8]. It can load and process heterogeneously structured data efficiently. Cylon was developed with the ideology that high-performance data processing should be available in a wide range of scenarios that involves a large amount of data, including data engineering, AI/ML applications, distributed databases, and so on. It is intentionally developed to be available as both a library that provides functions to load, extract, and transform data efficiently and as a framework that can run in a standalone fashion.

A Cylon dataframe represents one dataframe partitioned into a group of dataframes that may exist on different processes, machines, or clusters. The operators of such dataframes apply on each partition simultaneously with a Single Program Multiple Data (SPMD) pattern and use the collective operation to perform interaction between processes. Figure 1 presents an overview of Cylon’s multilayered architecture. Cylon consists of several data abstractions: dataframe/table, column, and scalar, and two levels of procedural abstractions: dataframe operators and communication operators. In this context, we refer to a column as an array of scalars of the same data type, and we refer to that data type as the data type of the column. We use the term dataframe interchangeably with table, as we refer to both of them as the data structure that consists of multiple columns of possibly different data types.

As Cylon is able to serve as a library, we envision it to be integrated with other data engineering or machine learning frameworks to help process data efficiently. For example, Ray is a distributed computing framework consisting of a distributed runtime and a machine learning library, able to scale and accelerate ML/AI applications. Another example is Dask, a distributed computing framework with task scheduling to scale Python applications. Both frameworks provide solutions to scale big data applications in Python effortlessly. Following Cylon’s ideology of making high-performance data processing available to every big-data scenario, we want to enable it to load and process data efficiently for applications using these frameworks. However, as Cylon uses MPI as its communication interface, we spot the incompatibility issue when trying to integrate Cylon with these frameworks.

We find UCX with UCC as an appropriate substitute for MPI. As an interface, MPI can build on many communication frameworks, including UCX. However, UCX can also operate as a communication framework itself. Like MPI, UCX and UCC are efficient and able to run on a variety of hardware and communication protocols. As UCX and UCC do not have their own bootstrapping mechanism, MPI is the de facto mechanism for them. To break this dependency, we implemented a more flexible initialization process. We modularized the bootstrapping process and decoupled it from any particular mechanism to enable the implementation of new mechanisms that specialize in certain environments.

In addition to being detached from any process-bootstrapping mechanism, UCX also brings benefits in performance and portability. UCX is decoupled from either specific network hardware or programming models, which brings great compatibility and portability without compromising performance or scalability.

Additionally, UCX provides seamless handling of Graphical Processor Unit (GPU) memory and full GPU-to-GPU direct communication, which makes it possible to accelerate applications further by GPU. Furthermore, Remote Direct Memory Access (RDMA) through InfiniBand and RDMA over Converged Ethernet (RoCE) is also supported by UCX, which enables some unique benefits in communication efficiency [9]. Firstly, UCX can make zero-copy GPU memory transfers over RDMA. Secondly, RDMA can significantly speed up overall communication. For example, with UCX over RDMA on the Spark GPU cluster, the NVIDIA Rapids plugin enables a 5$\times$ reduction in time for inventory pricing queries compared with the normal Spark GPU cluster[10]. Additionally, UCP is made to offer lightweight, portable interfaces over hardware abstraction layers and native network drivers. In order to support parallel programming models like Open-SHMEM, the UCP layer includes message layer features and protocols, including rendezvous protocols and tag matching for multi-rail networks [11].

In a nutshell, UCX is a very promising communication framework that brings a lot of critical modern features to high-performance distributed computing. By empowering Cylon with UCX/UCC, Cylon gains wider applicable scenarios and potential performance improvement for certain working environments.

We also considered Gloo[12] as an approach to our problem. However, UCX shows better performance results in our use case, as demonstrated in the experiment. In addition, as Gloo is built specifically for machine learning applications, UCX has a wider and more mature community in HPC.

The contributions described in this paper center around the Cylon communicator. The communicator is designed with the consideration that it will be implemented with multiple communication frameworks. As of now, we already have implementations with MPI, Gloo, and UCX/UCC. The communicator separates the data preparation process from the actual network-related code. Although it is effortless to achieve, not all data types are needed for each communication operator. Therefore, only selected communication operators are implemented for each data type. The availability of each data type to each communication operator is listed in table I.

Integrating UCX and UCC begins with implementing the Communicator and Channel. We use UCX to perform point-to-point operations in Channel and Barrier synchronization function and UCC to perform the collective operations. Different from other communicators such as MPI or Gloo communicator, as we use the two frameworks collaboratively with each other, we created one communicator, the UCX/UCC communicator, which embodies the functions of both of the frameworks.

When implementing the UCX/UCC communicator, we faced several challenges. The first challenge is about reusing resources. UC-Protocols (UCP) is the high-level API that UCX provides, and it can be accessed with a UCP endpoint, which contains all necessary resources for a particular network connection. Although UCC also uses the endpoints when implementing collective operations, there is no known way to reuse the endpoint we used in UCX on UCC. Therefore, we implemented the UCX/UCC communicator with two sets of UCP endpoints co-existing: one for implementing AllToAll with UCX and the other for the collective operations with UCC.

Another challenge is that, as of the time when we write this paper, UCC does not support Gather with variable data length in some circumstances. Therefore, we resembled the GatherV operation with the AllGatherV operation by allocating dummy space for receiving buffers in processes that are not gather-root. We hope that we can remove this temporary patch in the near future.

The process initialization includes collecting communication metadata required by the communication frameworks. Metadata, including world size and rank information, is utilized in many distributed computing frameworks, including UCX and UCC. In addition, each process also needs the endpoint’s address from every other process for communication to be possible. The world size, i.e., the number of processes running a specific job, is usually pre-defined and can be easily made available to each process. Therefore, we need to retrieve the rank information and the endpoint addresses for each process.

Strong scaling was initially measured for $10^{9}$ (1 Billion) rows (roughly 16GB in size), which are shown from Figure 6(a) to Figure 8(a). Ideally, all these operators should follow a linear graph because the work per worker reduces as we increase parallelism. All three communication implementations seem to follow this trend. However, UCX seems to be performing better than both OpenMPI and Gloo for larger parallelism.

We further analyzed this by looking at a strong scaling plot of $10^{8}$ (100 Million) rows (roughly 1.6GB in size) each. This would be predominantly communication-bound, as the data size is much smaller for computation. These plots are shown from Figure 6(b) to Figure 8(b). As expected, the communication performance disparity is more pronounced in this experiment, and UCX implementation shows much better scalability than the rest.

An important point to note here is that, currently, OpenMPI integrates UCX for internal communications [16]. Running our experiments using this integration would give a better idea about the communicator performance of Cylon.

To analyze the communication performance further, we carried out a weak scaling experiment for the join operation, with a communication and computation time breakdown. Times were measured for each hash-shuffle and the local join operation. The data set was generated with $5\times 10^{6}$ (5 Million) rows per relation per core. Therefore, when increasing from parallelism 32 to 512, the total data size would increase from $\sim$2.5GB to $\sim$40GB.

As expected, the local join time shows a nearly flat value. However, the shuffle time (i.e., the communication overhead) is increasing significantly along the parallelism axis. Gloo shows the worst performance, while the UCX communicator improves this deviation significantly. This justifies our decision to integrate UCX/UCC into Cylon. However, this experiment also shows that even though Cylon dataframe operators show decent performance, the current shuffle communication overhead might hinder the overall performance of Cylon in very large data sets and for large parallelisms. We expect to revisit the shuffle implementation with these insights as a future improvement.

There are a lot of existent data processing frameworks designed for efficient distributed workflow. For example, the MapReduce programming model[17] is one of the earliest and most famous ideas to accelerate data processing on large clusters by breaking down large data sets and processing them in parallel. Compared with MPI, the MapReduce framework can automatically parallelize user programs and provide transparent fault tolerance. MapReduce has provided a solid foundation for later successors, among which Spark[18] gradually dominates. As an in-memory processing framework, Spark can provide potentially a 10$\times$ speedup compared with MapReduce by avoiding writing back to disks during processing. Thanks to its high performance and low-latency data sharing, Spark is much more efficient with multi-pass applications like streaming processing, SQL, machine learning, etc. Although Spark benefits from its wide application scenes, it is not necessarily the best solution for some specific applications. Spark leverages micro batches to emulate streaming, which requires a careful trade-off between throughput and latency. Apache Flink[19] is yet another data processing framework that supports native streaming so that higher throughput and consistency are guaranteed. Additionally, Apache Flink has more support for iterative processing like machine learning by native loop operators which is absent in Spark.

Due to the rapid growth of dataset in a variety of field of data engineering, the use of heterogeneous data on single or multiple nodes push hard to extend its current limit. Dataframe(DF) and Distributed Dataframe(DDF) are the core unit of handling large-scale data engineering applications. Compared with the relational database table, DFs consist of homogeneous or heterogeneous data [20]. DF adaptation with heterogeneity distinguishes it from multidimensional arrays or tensors. DDFs solve the problem by acting as a unit placed in thousands of nodes concurrently and provides a common infrastructure on the cloud to handle billions of heterogeneous data for different ML or HPC applications and save significant development time (nearly 60%) [21]. In addition, Modin offers a pandas-like API that employs Ray or Dask to create a framework for high-performance distributed execution. On a computer with four physical cores, it offers speedups of up to four times. [22].

Partitioned datasets with well-separated blocks along an index can be effectively computed using the dask dataframe. Users gain from Dask dataframe’s straightforward access to larger-than-memory datasets and concurrent computing in cases where Pandas does release the Global Interpreter Lock (GIL) [23]. The higher-level collections dask dataframe for general computing shows the adaptability of the dask graph specification to encode complex parallel algorithms and the capacity of the dask schedulers to execute those graphs on a multi-core computer intelligently. Though it doesn’t neatly fit into a single high-level abstraction like arrays or dataframes and is instead merely a collection of related Python functions with data dependencies, dask graph allows for parallel execution to go beyond ndarrays and dataframes [1]. The dask runtime cannot yet be added to a distributed environment with a protocol-independent framework.

CuDF is a Python GPU DataFrame framework for reloading, joining, aggregating, filtering, and manipulating tabular data using a DataFrame style API. Where necessary, Dask-cuDF upgrades Dask to enable cuDF GPU DataFrames to process its DataFrame partitions rather than Pandas DataFrames. Dask-cuDF is advantageous if the workflow is spread across numerous GPUs, has more data than can fit in memory on a single GPU, or has to analyze data scattered across multiple files at once [24].

Compared with the frameworks mentioned above, Cylon empowered by UCX & UCC with distributed dataframe is designed for general high-performance computing environments.