Trends in High-Performance Data Engineering for Data Analytics

2.1 Data analytics frameworks

So far, we have discussed how data analytics have evolved and the role of data analytics frameworks. It is better to understand this deeply by looking into each framework. Among the data analytics frameworks, legacy machine learning frameworks like Scikit-learn [5] contained many statistical and machine learning models. But before moving into details of the most promising systems, it is worth mentioning that deep learning-based solutions have evolved way beyond the primary machine leanring models.

• PyTorch [6]: An open-source machine learning library based on Torch, which is used for applications such as natural language processing and computer vision. PyTorch is known for its dynamic computational graph, which allows for more flexibility in model creation and training. It is widely used in both academia and industry.

• TensorFlow [7]: A popular open-source framework for machine learning and deep learning that was developed by Google Brain. It supports a wide range of tasks, including image and speech recognition, and has a large and active community of users and contributors. TensorFlow’s key features include its data flow graph architecture and its ability to scale across multiple devices.

• MXNet [8]: A flexible and efficient deep learning framework that is known for its fast training speeds and low memory usage. Developed by Amazon Web Services, MXNet supports multiple programming languages and has a variety of pre-built models for image, text, and speech recognition.

• Caffe [9]: A deep learning framework that is focused on speed and scalability. It was developed by the Berkeley Vision and Learning Center and is known for its ease of use and powerful visualization tools. Caffe is commonly used for image classification, segmentation, and object detection.

• Theano [10]: A Python library that allows for efficient computation of mathematical expressions, particularly in the context of deep learning. It is known for its ability to optimize CPU and GPU usage and for its strong integration with NumPy. Theano is used in a variety of applications, including natural language processing and computer vision.

• Chainer [11]: A Python-based deep learning framework that was developed by Japanese company Preferred Networks. Chainer is known for its dynamic computational graph, which allows for more flexibility in model creation and training. It supports a wide range of tasks, including image and speech recognition, and has a variety of built-in optimization algorithms.

Even though frameworks like Tensorflow, MxNet and Theano are used by many data analysts and industrial work, it is worth noting that PyTorch has become the de-facto standard in the research community to develop deep learning models. The primary reason is it’s a configurable and wide array of APIs to break down the application development into finer details.

PyTorch is a machine-learning library that is built on top of the Torch library. It is designed to provide a user-friendly and flexible framework for building and training deep neural networks in Python. PyTorch offers support for dynamic computation graphs, which allows for modifying neural network architecture during runtime. PyTorch supports CPU, GPU, and TPU-based computation models, which are mainly designed to run on a bulk-synchronous-parallel communication model. In simple terms, AllReduce collective is regarded as the widely used communication operation when running distributed training programmes. In addition, PyTorch recently introduced an RPC-based computation model to provide more flexibility in running applications in cloud environments.

A wide variety of accelerators and communication models enables researchers to experiment with different model architectures and ideas without committing to a fixed architecture beforehand. It offers many built-in functions and classes for building and training neural networks, including modules for convolutional and recurrent neural networks, optimization algorithms, and loss functions. It also supports GPU acceleration, which enables efficient movement of deep neural networks on modern hardware. PyTorch has gained popularity in the machine learning community due to its ease of use, flexibility, and powerful capabilities. It is widely used in research and industry for various applications such as computer vision, natural language processing, speech recognition, and reinforcement learning.

2.2 Machine learning in production

Python provided an easier way to prototype applications, and later in production, these applications can be compiled into high-performance scripts using tools like Torchscript [12]. By decoupling the model from any runtime environment, TorchScript enables the model to be executed independently of the framework or platform it was developed on. This eliminates the Global Interpreter Lock (GIL) in Python, which can be a bottleneck for executing multithreaded inference. TorchScript prioritizes optimizing the entire program as a whole rather than just individual components or parts.

2.3 Computation intensity in data analytics problems

The performance required to run the most recent data analytics problems has exponentially evolved through the past decade. From image classifiers to intuitive Chess Players to intelligent chatbots, the cost of learning more has risen to new levels. Figure 1 [13] shows the computing cost for various deep learning models over time.

The data required to get an accurate model would depend on the model. Still, the modern-day argument in data-centric artificial intelligence [14] is that it is better to focus on a data-driven approach than a model-driven approach where the model is tuned to fit the data. This means that more and more data is required to get a better understanding, requiring data pre-processing at a larger scale.

Can we rely on the current data processing stack or enhance it? This is a significant question that needs to be addressed. How should the systems evolve? What are the best tools, or do we have to write new tools? How should we write such systems? All these questions are valid and very important. Let us learn more about data processing and how such systems have evolved in the past decade.

3.1 Data engineering frameworks

Large datasets cannot be efficiently processed with only a set of relational and linear algebra operators. As the amount of data generated every day continues to increase exponentially, reaching petabytes of data across various platforms serving millions of users, there is a need for a more versatile set of operators and operation modes. Many applications require real-time notifications, forecasts, instant messaging, and reporting within a specific timeframe. To support these requirements, data engineering frameworks encapsulate ways such as stream data processing and batch data processing. These frameworks provide the necessary infrastructure to process large datasets efficiently and effectively, making extracting valuable insights and information easier.

Such widely used data engineering frameworks are;

• Apache Hadoop: An open-source big data framework that provides distributed storage and processing of large data sets using a cluster of commodity hardware. Hadoop is used for batch processing, data analysis, machine learning, and more.

• Apache Spark [17]: A unified analytics engine for large-scale data processing that supports batch processing, streaming, and machine learning. Spark is designed to be fast and efficient, with in-memory computing and the ability to process data in parallel.

• Apache Kafka [18]: A distributed streaming platform that allows users to publish and subscribe to streams of records. Kafka is used for real-time data processing, data streaming, and data integration across different systems.

• Apache Flink [19]: A distributed stream processing engine that supports real-time stream processing and batch processing. Flink is designed to be highly scalable, fault-tolerant, and efficient, with support for both batch and stream processing in one system.

• Apache Beam [20]: An open-source, unified programming model for batch and streaming data processing. Beam provides a simple, consistent API for building data processing pipelines that can run on multiple processing engines, such as Apache Flink, Apache Spark, and Google Cloud Dataflow.

• Apache Storm [21]: a distributed stream processing framework that supports real-time processing of high-velocity data streams. Storm is used for real-time analytics, machine learning, and other applications that require fast and reliable data processing.

• Google Cloud Dataflow [22]: A fully managed, serverless data processing service that enables users to build batch and streaming data pipelines using Apache Beam programming model.

These frameworks are designed to solve a set of problems emerged through out the history of data processing. Table 1 shows the breakdown of pros and cons in these frameworks based on qualitative features like scalability, fault-tolerance and processing modes.

Apache Hadoop can be recognized as one of the earliest open-source big-data systems, and with time each system added its unique set of features. Apache software foundation and open-source software development significantly impact improving these systems. To see the progress made by these frameworks, we can look at Figure 2, which depicts the Github statistics.

Although these are the widely adopted data processing systems, researchers and engineers have made an enormous shift to gain much better performance. We start to see a pattern when looking into the programming languages used to develop the frameworks mentioned below (Table 2).

3.2 Data engineering evolution

Java and Scala are the programming languages used in developing these big-data systems. Even though Java is known for platform independence, security, and scalability, the performance aspect could be more pleasing for applications associated with heavy computing tasks. In that aspect, C++ or Rust are better replacements to gain much better performance.

Using high-performance languages like C++ and Rust has been one of the trends in modern large-scale data processing systems. After the big-data era, frameworks like Ray, Velox, DataFusion and Apache Arrow have been mostly used to design data processing pipelines. One key takeaway from these frameworks is how they provide usability. Unlike classical, big-data systems, which focused more on languages like Scala and Java to give the majority of their APIs, these frameworks focused on the Python programming language to a greater extent. Big-data systems provided Python wrappers known to slow performance due to serialization and deserialization issues when crossing language boundaries. Furthermore, the usability of these frameworks has been challenging due to their complex APIs. However, modern big data frameworks backed by C++ offer improved Python APIs with seamless integration with other data analytics systems.

Besides language preference, the widely used big-data systems like Apache Spark have been mainly designed to perform in a lazy execution model where a driver programme takes control of data partitioning and running tasks. This approach bottlenecks the aggregation or reduction tasks, which require all the spawned tasks to communicate with the driver programme to synthesize the final answer. This is a classical problem which can be seen in many big-data systems because of the scheduler semantics.

To address this issue, a few research efforts have been from the big data community to integrate Gang scheduling [23]. Apache Spark has also attempted to incorporate this concept into its schedulers. In addition, research frameworks like Twister [24], Twister2 [25] and Cylon [26] have introduced the usage of MPI for big-data processing by abstracting away the MPI collectives and providing a big-data like APIs for application developers. The performance gains are significant compared to the existing big-data systems.

3.3 Next generation data engineering frameworks

Next-generation data engineering frameworks were built to meet the requirements of modern-day data analytics systems. In Section 2, we learned a few key factors.

• More data leads to good results

• Data analytics systems run on accelerators

• Execution model is BSP or Asynchronous decentralized training (RPC)

• Python is the widely used programming language

• High performance is the key to efficient application development

Big-data systems have been there since 2005, and it has evolved in many ways to meet user requirements. But the major challenge came around 2012 when the data analytics world took over business modeling and solving scientific problems. The rise of neural networks in machine learning made shock waves through the entire data-driven echo system. This is where the aforementioned key factors come into play.

The motto of the last decade was the need to process more significant amounts of data to learn things better. And it began the evolution of modern-day data engineering systems. The key aspects that need an update are;

• Performance

• Usability

• Interoperability with Data Analytics systems

• Low learning curve

Big data systems can process more significant chunks of data but could improve further. The APIs provided in Java, and Scala could be more user-friendly for analysts to use in day-to-day work. There were some efforts to use Python (wrappers on Java/Scala APIs), but they could have been more user-friendly. The schedulers in classical, big-data and modern data analytics systems needed to be aligned, so running end-to-end pipelines was not easy. Also, providing an easier workflow for analysts to perform day-to-day tasks was important.

Figure 3 depicts an approximate performance estimation compared to the usability. Note that this is not a mathematical outcome based on experiments but a collective approximation based on the evolution based on timeline and experience. There is always room for improvement; engineers find ways to improve systems.

Around 2016, a few projects started focusing on better performance and usability. These projects realized the underneath issues in the existing data processing systems. Each project addressed various aspects of the data processing domain, but collectively they can offer a much better solution to data-driven applications.

Figure 4 depicts the main aspects where data engineering frameworks have been evolving. Distributed computing is not a new aspect of data engineering; the older frameworks can do distributed computing. But frameworks like Ray and Cylon have the edge over the existing distributed computing approaches for two main reasons. These two systems have a C++ core backing the performance of sequential operators and communication operations.

Although, before the time of high-performance computing engines, there was an invention called Dask [27] which was mainly designed to provide a set of operators on Pandas [28] to provide distributed computing on a primary workstation. These frameworks were seamlessly integrated with Pandas and Numpy, which made it easier to work with deep learning and machine learning libraries. It later evolved into a distributed framework even to scale in supercomputers. Dask-Distributed is a distributed computing framework that is designed to enable efficient processing of large-scale data sets in Dask with Python. It is built on top of standard Python libraries, such as NumPy [29], Pandas, and Scikit-Learn, and provides a flexible programming model that allows users to write distributed applications with ease. Dask also offers a range of distributed data structures, including arrays, data frames, and bags, which can be used to represent and manipulate large-scale data sets in a distributed environment. These data structures are designed to be familiar to users of NumPy, Pandas, and other Python libraries, making it easy to work with large data sets in a distributed setting. To distribute computations across multiple nodes, Dask uses a task scheduler, which enables users to schedule and manage analyses across a cluster of machines. The scheduler is designed to be fault-tolerant, ensuring that computations continue to run even if some nodes fail or become unavailable. Dask also includes a range of performance optimizations, such as data partitioning and compression, to ensure that computations are completed as efficiently as possible. But one of the critical challenges in the Dask system is it is entirely developed on Python. When running larger workloads, Dask tends to decline performance. The main reasons are the less performance from the Python language and bottlenecks with GIL when running compute-intensive workloads.Listing 1.1.

Example Dask Join [30].

1 import dask

2 large = dask.datasets.timeseries(freq=“10s”, npartitions=10)

3 small = dask.datasets.timeseries(freq=“1D”, dtypes=(“z”: int))

4 

5 small = small.repartition(npartitions=1)

6 result = large.merge(small, how=“left”, on=[“timestamp”])

A framework called Ray was introduced a few years back to provide an abstraction on distributed training for reinforcement learning and deep learning. Ray provides a flexible programming model that allows developers to write distributed applications in Python with minimal effort. Ray provides an actor based compute model which is easier to scale. It is built on top of the Apache Arrow data format, enabling efficient data transmission without serialization and deserialization between different programming languages or services. It also offers a distributed task scheduler, allowing the users to schedule and manage complex workflows across multiple nodes. Ray is optimized for machine learning applications and provides several built-in libraries and tools for developing and deploying ML models at scale. Overall, Ray aims to make it easy for developers to build and scale distributed applications without worrying about the underlying infrastructure. A commercial version of Ray allows users to work with cloud environments and design applications quite efficiently. An autoscale feature will enable developers to parallelize workloads with trivial command-line arguments. Both Ray and Dask are cloud-friendly tools. But there are systems that were invented way before the cloud was created and are very fast compared to big-data processing systems. These systems are built on high-performance computing (HPC) libraries like MPI [31], OpenMP [32], PGAS [33].Listing 1.2.

Example Ray code for data processing [34].

1 import ray

2 import pandas

3 

4 ds = ray.data.read_csv(“/path-to-iris-data”)

5 ds.show(3)

6 # Repartition the dataset to 5 blocks.

7 ds = ds.repartition(5)

8 # Find rows with sepal.length < 5.5 and petal.length > 3.5.

9 def transform_batch(df: pandas.DataFrame) -> pandas.DataFrame:

10 return df[(df[“sepal.length”] < 5.5) (df[“petal.length”] > 3.5)]

11 # Map processing the dataset.

12 ds.map_batches(transform_batch).show()

13 # Split the dataset into 2 datasets

14 ds.split(2)

15 # Sort the dataset by sepal.length.

16 ds = ds.sort(“sepal.length”)

17 ds.show(3)

18 # Shuffle the dataset.

19 ds = ds.random_shuffle()

20 ds.show(3)

21 # Group by the variety.

22 ds.groupby(“variety”).count().show()

With an HPC-oriented approach, a framework called Cylon has been specially designed on an MPI-backed collective communication model, which provides high-performance computing capability on Supercomputers. Cylon core communication and compute kernels are written in C++ and extended on Apache Arrow data structures to represent data efficiently. This allows the ability to seamlessly integrate with other Arrow-backed systems and provide efficient data movement from data engineering frameworks to machine learning and deep learning workloads [35, 36, 37]. The data structure used underneath is Apache Arrow. It uses Arrow compute kernels to do sequential relational algebra operations while using its partitioned API written on MPI to corresponding distributed operators. Cylon provides APIs in C++, Python and Java. But Cylon has focused on providing more support to Python users by providing a distributed DataFrame library which mimics Pandas but provides distributed operators which abstract away the complex communication algorithms. Cylon supports both CPU and GPU computing. As Apache Arrow for sequential computes, it uses CuDF [38] to do the sequential operators and uses a GPU-supported partition algorithm for the distributed operators. Cylon can be recognized as one of the earliest distributed DataFrame libraries supporting GPU and CPUs. Although Cylon is developed on HPC-based communication models, recently, it has adopted UCX and Gloo as communication backends to enable the workloads to run in cloud-native environments [39].

Big-data frameworks like Apache Spark, Apache Storm, Apache Flink and similar other frameworks have a downside when it comes to being a perfect match for data engineering for data analytics. The gap it left in data engineering is that it is hard to integrate them with the HPC-Pythonic data analytics stack, which runs on HPC hardware with a C++ core backend and an easy-to-use Python API. Cylon fills this gap by providing these high-performance communications and computing APIs to work with data. And enhancing this experience, DataFrames, which Pandas introduced, has become the bedrock of usability. This is one of the unique features of Cylon. Looking into the PyCylon code to do a join in listing 1.3, it is clear that the API is very similar to what Pandas is offering. Pandas has become the go-to tool for data processing for analytic data workloads. But Pandas is a sequential library. Because data analytics frameworks like PyTorch run on BSP-model, they enable MPI-enabled Cylon Dataframes to run seamlessly with data analytics workloads. It is vital when the exploratory analysis is done before designing the production-ready model, where data scientists play with the data to engineer the features and get feedback with the analytical algorithms. It is vital to note that introducing Pythonic HPC solutions is crucial in designing efficient data exploration research to build larger models.Listing 1.3.

Example Python code with pycylon.

1 from pycylon import read_csv, DataFrame, CylonEnv

2 frompycylon.netimport MPIConfig

3 

4 config: MPIConfig = MPIConfig()

5 env: CylonEnv = CylonEnv(config=config, distributed=True)

6 

7 df1: DataFrame = read_csv(’/tmp/csv1.csv’)

8 df2: DataFrame = read_csv(’/tmp/csv2.csv’)

9 

10 df3: DataFrame = df1.join(other=df2, on=[0], algorithm=“hash”, env=env)

11 

12 print(df3)

Both Ray and Cylon provide much better performance compared to Dask. The main reason is the efficiency of computing kernels written on C++ and distributed computing models. Dask also uses a driver-centric distributed computing model, which becomes a bottleneck in running more tasks that need more synchronization than a job with fewer tasks.

Numerous libraries, including Ray and Cylon, utilize Apache Arrow as their underlying data structure. Apache Arrow is a columnar in-memory data format that enables efficient read operations. Its core memory layout is based on a Columnar specification. It allows any framework to adopt the Arrow C Data Interface and extend it to create Arrow-compatible data structures without relying on the entire library. Several libraries have been built upon this columnar specification, including the compute API, dataset API, Flight SQL, Flight RPC, and Acero streaming execution engine. Apache Arrow has gained widespread adoption in various industrial frameworks and academic research, such as Apache Spark, Clickhouse, Dremio, and Polars. Moreover, it supports an extensive range of programming languages, with C++, Python, R, Java, C#, and Go being the most commonly used.

In the heart of data processing, there lie the query engines. Query engines are known to provide a higher-level API for users to run SQL queries or build query plans based on an API. A few key goals of a query engine are fast query execution, scalability, flexibility, concurrency, query optimization, usability, fault tolerance and extensibility. It should have a way to efficiently load, transform, persist and transmit across a wide array of other systems. Modern-day query engines are not built on Java or Scala like in the big-data era; they are built on C++ or Rust for performance requirements. Velox [40] is a high-performing query engine built on C++. It uses Apache Arrow columnar format and has its own set of compute kernels. Velox also supports Presto and Apache Spark query engines. In addition, it also supports feature engineering and data preprocessing in PyTorch. It is a novel system evolving in the high-performance query processing space. Velox is getting more traction in industrial and academic research.

DataFusion [41] is a robust data processing framework that provides users with two distinct methods for creating logical query plans: SQL and DataFrame API. This versatility enables users to choose the most suitable approach for their specific use cases and requirements. The framework also boasts a comprehensive query optimizer that employs advanced optimization techniques to improve the efficiency and performance of query execution. By analyzing and transforming the logical query plans, the optimizer ensures that the most effective execution strategies are used, resulting in faster processing times and better resource utilization. A key feature of DataFusion is its multi-threaded parallel execution engine, designed to process partitioned data sources, such as CSV and Parquet files, with exceptional speed. By using parallelism, the execution engine can distribute the workload across multiple threads or cores, significantly accelerating data processing and analysis. This parallel execution approach is particularly beneficial when working with large datasets, as it can effectively minimize processing times and overcome performance bottlenecks. In summary, DataFusion combines the flexibility of SQL and DataFrame API support, an advanced query optimizer, and a high-performance parallel execution engine to offer a robust and efficient solution for processing and analyzing partitioned data sources like CSV and Parquet files. Its versatile and powerful capabilities make it ideal for various data processing tasks and use cases.

Considering the discussed novel trends and data processing technologies, it is evident that data engineering has evolved from one dimension to another in less than a decade. Researchers and engineers have produced various enhancements to the data engineering stack, and high performance and better usability are the key aspects that have shown progress. Moreover, the number of novel platforms trying toimprove data engineering workloads gives a variety of options for the user. But most of these frameworks have focused on two essential things. Represent and transform data efficiently and seamlessly integrate with the data analytics workloads.

3.4 Anywhere query execution

Among hundreds of frameworks designed for data engineering with uniqueness, each framework is known to be better at specific tasks than the others. In a practical scenario, a few frameworks form data engineering workflows. The main challenge is communicating an idea or simply a query plan so that each framework can understand and do its part. Apache Beam can be recognized as a single framework which unifies frameworks like Apache Spark, Apache Flink, Apache Samza, Google Cloud Dataflow, Twister2, etc. This approach requires writing composite applications by using the Apache Beam API. But in the long run, maintaining such a code base and supporting various platforms is complex and costly. We must run your query anywhere with less overhead and maintenance.

Substrait [42] provides a cross-language specification for data computing operations. In simple terms, once a framework adopts the Substrait specification, it can run a query plan without involving any additional code except for the code required to load a Substrait-based execution plan and execute the framework-native plan. Substrait currently support types, expressions and relations. Under types, data types, type variations and functions (scalar, aggregate, window and Table) are defined. If a particular framework wants to extend upon the existing definitions, such modifications can be done with Substrait. Especially when defining new data types, function signatures and other custom representations of vivid components. These are defined as extensions, which can be defined in a YAML format [43].

At the query execution level, the most exciting component is the Relational algebra support in Substrait. It contains logical and physical relations defined to support most of the widely used relational algebra operators. Read, filter, sort, project, join, set, aggregate, and write are supported, operators. To understand what Substrait can offer, let us evaluate a sample query and how Substrait can represent it. Listing 1.4 shows a SQL query which performs a read operation on a table LINEITEM and reading L_EXTENDEDPRICE, L_TAX and L_DISCOUNT columns.Listing 1.4.

Example of a Substrait plan in SQL Format.

1 SELECT

2   t0.”L_EXTENDEDPRICE”,

3   t0.”L_TAX”,

4   t0.”L_DISCOUNT”

5 FROM “LINEITEM” AS t0

Substrait does not have a visual format at the moment. It is a protobuf-based non-human readable format which can be sent to Substrait supported framework to execute using the native query engine. But to visualize it with particular readability, we show it using the JSON format. Listing 1.5 offers the Substrait plan in Visual form.Listing 1.5.

Example of a Substrait plan in JSON Format.

1 {

2   “extensionUris”: [

3     {

4       “extensionUriAnchor”: 1

5     }

6   ],

7   “relations”: [

8     {

9       “root”: {

10         “input”: {

11         “read”: {

12          “common”: {

13           “direct”: {}

14          },

15          “baseSchema”: {

16           “names”: [

17            “L_EXTENDEDPRICE”,

18            “L_TAX”,

19            “L_DISCOUNT”

20          ],

21          “struct”: {

22           “types”: [

23            {

24             “fp32”: {

25            “nullability”: “NULLABILITY_NULLABLE”

26              }

27            },

28            {

29             “fp32”: {

30             “nullability”: “NULLABILITY_NULLABLE”

31              }

32            },

33            {

34             “fp32”: {

35             “nullability”: “NULLABILITY_NULLABLE”

36                }

37              }

38            ],

39             “nullability”: “NULLABILITY_REQUIRED”

40           }

41          },

42           “namedTable”: {

43               “names”: [

44                “LINEITEM”

45              ]

46            }

47          }

48        },

49        “names”: [

50           “L_EXTENDEDPRICE”,

51           “L_TAX”,

52           “L_DISCOUNT”

53        ]

54      }

55    }

56  ]

57 }

The Substrait plan contains information about the data source, a namedTable representing an in-memory data source. In addition, a file source or a glob can be referred to via URI to define the data source. Since this is a simple read operation, it only shows the baseSchema (schema of the data being read, not the schema of the dataset represented in the data source). One of the doubtful questions which arise is why not SQL? SQL is a language used for querying relational data, but it has limitations and lacks sufficient detail for processing. Therefore, modern systems often translate SQL queries into a query plan before executing them. Query plans can have multiple levels and transform, but no standard or open format exists for them. Substrait was created to provide a standard and available form for query plans and works alongside SQL to deliver capabilities that SQL lacks.

Acero (Arrow streaming query engine) [44], Velox, DataFusion and Ibis support Substrait. In terms of support, there are two aspects. There must be a set of producers who can produce Substrait plans and consumers who can execute them. Ibis supports the production and consumption of Substrait plans, while Velox, DataFusion and Acero mainly support Substrait plan consumption. But regarding a producer, the goto tool is isthmus [45].