Large-scale computing frameworks are key technologies to fulfill the computational requirements of massive data analysis. In particular, while Apache Spark has emerged as de facto standard for big data analytics after Hadoop's MapReduce, tools such as Dask and Ray can greatly boost the performance of Python applications in distributed environments. The goal of this paper is to study the performance of these three frameworks on a common playground. We focus on cloud-native architectures, which merge the benefits of big data and cloud computing. We refrain from considering high-level features such as ML models, we instead consider simple data processing operations, common ingredients of more complex pipelines. As a byproduct of our experiments, we ofer a set of guidelines for the development of cloud-native data processing applications.
CEUR ceur-ws.org
Massive data processing pipelines running on single-node machines or large clusters can benefit
of computing frameworks that hide low-level operations needed to parallelize the workload.
In this regard, so-called Big Data engines play a crucial role as they make it simple to write
applications that transparently scale with the underlying infrastructure. For instance, Apache
Spark1 has emerged as the de facto standard for big data analytics after Hadoop’s MapReduce [
The growth of these technologies has been accompanied by the growth of cloud-native
applications, which are increasingly studied in research and adopted in industry. Cloud-native
applications consists of several processes that run in isolated containers, which are typically
spread over the nodes of a cluster. In this respect, Kubernetes4 is becoming the standard
technology to automate software deployment, scaling, and management through containers
orchestration [
Given the popularity of Spark, Dask, and Ray, and the importance of making informed
choices in terms of technology, the goal of this paper is to study the performance of these
three frameworks on a common playground. We use a cloud-native environment based on
Kubernetes, and we experiment standard data processing pipelines fed by a large dataset
consisting of numerical values. In particular, while we are well aware of the diferences and
peculiarities in terms of high-level APIs ofered by the considered frameworks, our focus is on
standard preprocessing operations that are typically performed in any data-processing pipeline.
We do not consider high-level features such as ML/DL models as they require diverse datasets
and specialized hardware (such as suitable GPUs), to be conveniently evaluated. As a byproduct
of our experiments, we ofer a set of guidelines for the development of cloud-native
dataprocessing applications. To the best of our knowledge, this is the first paper that experimentally
compares these three frameworks. The only similar comparison we are aware of is restricted to
Dask and Spark, and it focuses on neuroimaging pipelines [
Paper structure. A detailed description of the experimental setting can be found in Section 2. Results and guidelines are discussed in Section 3 and Section 4, respectively. Section 5 concludes the paper with future research directions.
In this section we describe the infrastructure and the benchmark used to run the experiments. Hardware. In terms of hardware, the experiments were conducted on a Dell PowerEdge XE8545 server, designed to take advantage of the industry’s most advanced technologies. The server is equipped with two sockets and has a 4U format, with two 128-core AMD EPYC processors, NVIDIA A100 Tensor Core GPUs and 1024 GB of RAM. As for storage, it has 4 Dell Ent NVMe CM6 disks of 2TB each. See also Table 1 for full references. Furthermore, it is equipped with NVLink, PCIe Gen 4.0 and NVMe SSD to improve I/O and data processing performance. These technologies speed up data transfer between the CPU and GPU, improving
the overall eficiency of the system. The server is certified by NVIDIA for its performance, manageability, security and scalability.
Processor Memory
GPUs Operating System
Storage
Dual AMD EPYC 7713, 64C, 2.8 GHz 1024 GB (16x64GB @ 3200 MT/s) 4 x NVIDIA A100 SXM4 80 GB
Linux Ubuntu Server 20.04 LTS 4 x Dell Ent NVMe CM6 RI 1.92TB 2.1.8
Kubernetes. We deployed a bare-metal single-node Kubernetes cluster on the previously
described server. We remark that the scheduling and resource allocation process is a critical
aspect for such experiments [
Listing 1: Volcano scheduler configuration. 1 a p i V e r s i o n : v1 2 d a t e : 3 v o l c a n o − s c h e d u l e r . c o n f : | 4 a c t i o n s : ” enqueue , a l l o c a t e , b a c k f i l l ” 5 t i e r s : 6 − p l u g i n s : 7 − name : p r i o r i t y 8 − name : gang 9 − name : c o m p l i a n c e 10 k i n d : ConfigMap 11 m e t a d a t a : 12 c r e a t i o n T i m e s t a m p : ” 2 0 2 3 − 0 2 − 1 5 T09 : 1 6 : 4 7 Z ” 13 name : v o l c a n o − s c h e d u l e r − c o n f i g m a p 14 namespace : v o l c a n o − s y s t e m
For each framework a specific queue has been defined, as reported in Listing 2, in order to decouple the workloads of the diferent frameworks; see also Fig. 1 for a schematic illustration.
Listing 2: Volcano queue definition. 1 a p i V e r s i o n : s c h e d u l i n g . v o l c a n o . sh / v 1 b e t a 1 2 k i n d : Queue 3 m e t a d a t a : 4 name : t e s t − s p a r k
Finally, to ensure reproducibility, Table 2 provides all the versions and configurations used for the entire software stack.
Storage. Read and write operations are fundamental in the context of data processing. Therefore, it is essential to configure the system to operate in the best conditions. In particular, we implemented a scalabale object storage system via MinIO7 within the Kubernetes cluster. This system consists of 4 physical disks for a total of 7 TiB of available space.
Dataset. The dataset was generated synthetically by sampling random values in the
range [
Format. As for the format in which to store the dataset, we used Apache Parquet [
Listing 3: Definition of a Spark Application. cpu : 32
Metrics and procedure. In order to assess the performance of the experimented frameworks, besides the runtime, we considered both speedup and eficiency, briefly recalled below. Let be the number of available processors, let be the time taken to process the workload with ∈ [1, ]
processors, then the metrics are defined as follows (a linear speedup and constant
eficiency are ideal targets, see, e.g., [
Each framework has been evaluated with a number of workers equal to 2 , with ranging in
[
Overall performance.
We being by remarking that, given a pipeline of operations, each framework adopts a specific strategy to break the workload into smaller tasks and to assign the (1) (2) speedup () = eficiency () = 1 ()
tasks to the workers. In particular, triggering the execution of single operations to measure each single run time would hinder such optimization strategies and invalidate the results. Therefore, we measure the run time of the whole pipeline, such that each framework can fully exploit its own optimization strategy. The run time includes the time needed to read the data, which will also be evaluated separately in the next paragraph.
Table 4 reports the results of the experiments. In terms of overall execution time, Ray appears to be the fastest framework, with the best performance in all worker configurations except in the single worker setting, in which Dask is only a few seconds faster. Between Dask and Spark, we see that Dask outperforms Spark up to 2 workers, while Spark performs better between 4 and 32 workers. In terms of speedup and eficiency, Spark exhibits very good performance up to 16 workers, showing an eficiency drop with 32 workers. The Ray framework shows better scalability, with a smaller drop in terms of eficiency. On the other hand, Dask exhibits low scalability compared to the other two frameworks, its eficiency rapidly decreases as the number of workers increases.
Reading performance. In this paragraph we analyze the performance of the frameworks in terms of reading performance, that is, the time taken to read the data from the object storage and to load the data into the main memory. Table 5 reports the results of our experiments. Again, for the single worker scenario, Dask is the fastest framework, and it remains the fastest also with 2 workers. With 4 workers Spark becomes the fastest framework, and it remains the fastest also with 8 workers, together with Ray. With 16 and 32 workers, Ray is the fastest framework, with Spark slightly slower. In terms of speedup, again Ray and Spark outperform Dask, which also exhibits the worse behavior in terms of eficiency.
Discussion. We begin by briefly discussing possible relations between the design principles
behind the three frameworks and the observed performance. Spark has been designed to handle
large amounts of data and complex operations, using distributed data structures to keep data in
memory and minimize disk access operations. In particular, its ability to optimize operations
through its distributed execution engine, dividing the operations to be performed into tasks and
distributing them eficiently on the nodes of the cluster, makes it able to scale very well with
configurations with high availability of resources, while it appears less eficient with few workers.
Ray is a system designed to handle high-speed, low-latency distributed processing operations.
To achieve this goal, eficient communication between nodes and an actor-based programming
model are used, which allows operations to be distributed asynchronously, minimizing workers’
downtime. This architecture makes it very eficient in all configurations. Dask uses a data
partitioning strategy based on blocks of variable size, which allows to better adapt to the size of
the dataset and to minimize data movements between nodes. With few workers, Dask’s data
partitioning strategy proves to be particularly efective, as it allows you to make the most of
the available resources, minimizing execution times. However, as the number of worker nodes
increases, coordinating distributed operations becomes more complex, and Dask may show
limits in terms of scalability compared to Spark and Ray. In particular, Dask can sufer from
increased overhead for coordinating distributed operations and managing distributed memory,
which can slow performance on large clusters.
Limitations. Our experiments where performed on a highly optimized infrastructure,
however, the performance of the frameworks are afected by the use of containers. For instance,
it was shown that Kubernetes pods may deteriorate data locality, and make a worse usage of
memory and CPU [
Guidelines. Based on our findings and with the above limitations in mind, Ray appears to be the most eficient framework over all configurations. On the other hand, Dask is the fastest framework with only one worker, but it is less scalable than Spark and Ray. Spark performs somewhere in between, with performance improving as nodes increase. In the case of read-intensive applications, Dask seems a good choice if only few nodes are available, while Spark or Ray would be recommended with a large number of workers. Besides performance, the ecosystem is an important aspect to consider when choosing a technology over another. In this respect, Spark stands out for its reliability, as it is certainly a well-tested framework, with an established community of developers ofering a wide range of resources, including documentation and reusable code. On the other hand, Ray allows you to run Spark and Dask code on the infrastructure managed by the Ray Core, thus making it possible to use particular features of other frameworks or existing legacy code. Another key aspect when staring out a project with a new technology is the required learning curve. In this regard, Dask’s learning curve is rather shallow for Python developers, with the possibility of using the same syntax of the most common libraries, such as Pandas and Numpy.
We plan to extend our experiments by considering larger datasets and more advanced features ofered by the three frameworks. In particular, we would like to compare their performance in terms of model training. To this aim, we plan to equip our cluster with suitable GPUs, which represent the current standard technology in this field. This type of comparison should involve diferent ML/DL models and hence multiple datasets, for instance, including images and text.