Single Core to Multi-Core CPUs

Although this is a book about data science, sometimes it is necessary (and nostalgic) to take a slight detour into the hardware realm and revisit the history of development on that side. For parallel computing, a lot of hardware development had to happen over a long period of time before the modern software stack started taking full-blown advantage of that development. It will be beneficial to get a brief glimpse of this history to put our discussion in context.

The earliest commercially available CPU was the Intel 4004, a 4-bit 750kHz processor released in 1971. Since then, processor performance improvements were mainly due to clock frequency increases and data/address bus width expansion. A watershed moment was the release of the Intel 8086 in 1979 with a max clock frequency of 10MHz and a 16-bit data width and 20-bit address width.

The first hint of parallelism came with the first pipelined CPU design for the Intel i386 (80386) which allowed running multiple instructions in parallel. Separating the instruction execution flow into distinct stages was the key innovation here. As one instruction was being executed in one stage, other instructions could be executed in the other stages and that led to some degree of parallelism.

At around the same time, superscalar architecture was introduced. In a sense, this can be thought of as the precursor to the multi-core design of the future. This architecture duplicated some instruction execution units, allowing the CPU to run multiple instructions at the same time if there were no dependencies in the instructions. The earliest commercial CPUs with this architecture included the Intel i960CA, AMD 29000 series, and Motorola MC88100.

The unstoppable march of Moore’s Law (shrinking the transistor sizes and manufacturing cost at an exponential pace; www.synopsys.com/glossary/what-is-moores-law.html) helped fuel this whole revolution in microarchitecture with the necessary steam. Semiconductor process technology was improving and lithography was driving the transistor nodes to the realm of sub 100 nm (1/1000^th of the width of a typical human hair), supporting circuitry, motherboards, and memory technology (taking advantage of the same manufacturing process advancements).

The war for clock frequency heated up and AMD released the Athlon CPU, hitting the 1GHz speed for the first time, at the turn of the century in 1999. This war was eventually won by Intel, who released a dizzying array of high-frequency single-core CPUs in the early 2000s, culminating in the Pentium-4 with a base frequency around 3.8GHz - 4GHz.

But fundamental physics struck back. High clock frequencies and nanoscale transistor sizes resulted in faster circuit operations, but the power consumption shot up as well. The direct relationship between frequency and power dissipation posed an insurmountable problem for scaling up. Effectively, this power dissipation resulted in so-called higher leakage current that destabilized the entire CPU and system operation when the transistor count was also going up into billions (imagine billions of tiny and unpredictable current leakages happening inside your CPU).

To solve this issue, multi-CPU designs were tried that housed two physical CPUs sharing a bus and a common memory pool on a motherboard. The fundamental idea is to stop increasing the frequency of operations and go parallel by distributing the computing tasks over many equally powerful computation units (and then accumulate the result somehow). But due to communication latencies from sharing external (outside the package) bus and memory, they were not meant to be truly scalable designs.

Fortunately, true multi-core designs followed soon after where multiple CPU cores were designed from the ground up within the same package, with special consideration for parallel memory and bus access. They also featured shared caches that are separate from the individual CPU caches (L1/L2/L3) to improve inter-core communication by decreasing latency significantly. In 2001, IBM released Power4, which can be considered the first multi-core CPU, although the real pace of innovation and release cycle picked up after Intel’s 2005 release of the Core-2 Duo and AMD’s Athlon X2 series.

Many architectural innovations and design optimizations are still ongoing in this race. Enhancing core counts per generation has been the mainstay for both industry heavyweights, Intel and AMD. While today’s desktop workstation/laptops regularly use 4 or 6 core CPUs, high-end systems (enterprise data center machines or somewhat expensive cloud instances) may feature 12 or 16 cores per CPU.

For the data science revolution and progress, it makes sense to follow this journey closely and reap the benefits of all the innovations that hardware design can offer. But it is easier said than done. Parallelizing everyday data science tasks is a non-trivial task and needs special attention and investment.

What Is Parallel in Data Science?

For data science jobs, both data and models are important artifacts. Therefore, one of the first considerations to be made for any parallel computing effort is to the focal point of parallelizing: data or model.

Why do you need to think this through? Because some artifacts are easier to be imagined to be parallelized than the others. For example, assume you have 100 datasets to run some statistical testing on and 4 CPU cores in your laptop. It is not hard to imagine that it would be great if somehow you could distribute the datasets evenly across all the cores and execute the same code in parallel (Figure 10-1). This should reduce the overall time to execute the statistical testing code significantly, even if the scheme involves some upfront overhead for dividing and distributing data, and some end-of-the-cycle aggregation or accumulation of the processed data.

Although this is not hard to imagine, the actual implementation is not that straightforward for a traditional Python-based data science stack using pandas or SciPy. As discussed in the last chapter, Python is inherently single-threaded and doing parallel processing with Python code needs some prior setup and clever manipulation. When a data scientist is using high-level analytics libraries like pandas, it is even more important to know the limitations for parallel processing (if any).

Moreover, data science is not limited to model exploration and statistical analysis on a single person’s laptop (or a single cloud-based compute node) anymore. From a single machine (however powerful it might be), the advantages are apparent for large-scale data analytics when one connects to a cluster architecture consisting of multiple computers banded together with high-speed network. In the limiting case, such a cluster aims to become a single entity of computing for all intents and purposes: a single brain arising out of parallel combination and communication among many smaller brains.

Naturally, data scientists start imagining all kinds of possibilities that can be tried and tested with this collective brain. Alongside splitting a large collection of datasets, they can think of splitting models (or even modeling subtasks) into chunks and executing them in a parallel fashion. Datasets can be sliced and diced in multiple ways and all those dimensions might be parallelized, depending on the problem at hand. Some tasks may benefit from splitting data samples in rows; others may benefit from column-wise splitting.

Even many optimization tasks can be parallelized with sufficient effort and thrown to multiple compute nodes. One example could be running parallel local area searches for finding the best cost function of a global problem. All these ideas are captured in Figure 10-2.

How Dask Works Under the Hood

At its core, Dask operates by using efficient data structures (arrays and DataFrames) and a cleverly designed graph. Basically, it uses a client-scheduler-worker cluster architecture (Figure 10-3) to optimally distribute subtasks, collect them together, and calculate the outcome/prediction. The intricacies of parallel computing are abstracted away from regular Python programmers or data scientists, so working with large datasets is made easy and accessible. Figure 10-3 shows a schematic illustration.

Dask DataFrame

Essentially, a Dask DataFrame is a large-scale parallelized DataFrame composed of many smaller pandas DataFrames, split along the index. Depending on the size and situation, the pandas DataFrames may exist on the disk for out-of-core computing on a single machine, or they may live on many different computing nodes in a cluster. A single Dask DataFrame operation triggers many operations down the chain (i.e., on the constituent pandas DataFrames in a parallel manner).

Efficiency and ease of use are main goals of the Dask project. Therefore, Dask DataFrames are partitioned row-wise, grouping rows by index value for efficiency. At the same time, they can expose the same API and methods as those coming from the pandas stable. A data scientist won’t feel the difference or need to change existing code but can utilize the parallelism just by working with the Dask DataFrame API. In fact, the pandas official documentation suggests using Dask for scaling out to large datasets (Figure 10-4).

Dask Task Graph

Dask uses the common approach to parallel execution in user-space: task scheduling. With this approach, it breaks the main high-level program/code into many medium-sized tasks or units of computation (e.g., a single function calls on a non-trivial amount of data). These tasks are represented as nodes in a graph. Edges run between nodes if one task is dependent on the data produced by another. A task scheduler is called upon to execute this whole graph in a way that respects all the inter-node data dependencies and leverages parallelism wherever possible, thereby speeding up the overall computation.

There are many techniques for scheduling: Embarrassingly Parallel, MapReduce, Full Task Scheduling, etc. Often task scheduling logic hides within other larger frameworks like Luigi, Storm, Spark, and IPython Parallel. Dask encodes full task scheduling (Figure 10-5) with minimal incidental complexity using common Python artifacts (i.e., dictionaries, tuples, and callables). Dask can even use Python-native schedulers such as Threaded and Multiprocessing.

Taking the fundamental data structures and schedulers, we can illustrate the flexibility of Dask as shown in Figure 10-6.

Basic Usage Examples

Here is how you can define and examine some of the data structures you just learned about. Let’s start with arrays and then go on to show some examples with DataFrames and Bags.

Array

Define a Numpy array of 100,000 elements (Gaussian random numbers) and create a Dask array from that using the da.from_array() method:

import numpy as np

import pandas as pd

import dask.dataframe as dd

import dask.array as da

import dask.bag as db

arr = np.random.normal(size=100_000).reshape(500,200)

dask_arr = da.from_array(arr,chunks=(100,100))

Note that for creating the Dask array, you have chosen a chunk size of (100, 100). In a Jupyter notebook, if you just examine this dask_arr object, it is even visualized nicely (Figure 10-7).

All the chunks have the same size of 78.12 kiB whereas the total dataset is 781.25 kiB. These chunks can effectively be distributed over cores or machines for parallel computing. You can go ahead and define a 3D array in a similar fashion:

arr = np.random.normal(size=100_000).reshape(50,200,10)

dask_arr = da.from_array(arr,chunks=(50,20,10))

Now the Dask array looks like a stack of bricks with a 3D shape (Figure 10-8).

Dask operates on the principle of lazy valuation where final values are not computed unless explicitly asked to do so. You can define a summation operation on the 3D array like this where you are also counting the time for the operation:

import time

t1 = time.time()

task1 = dask_arr.sum(axis=2)

t2 = time.time()

print("Time (milliseocnds):", round((t2-t1)*1000,3))

task1

In the Jupyter notebook, this will show a visualization. Note the time taken for this operation is ~4 milliseconds (Figure 10-9). Nothing has been computed; just a task graph has been built, and the expected output array shape has been determined.

Similarly, you can add another operation to this chain, determining the max value out of those summed values along the columns (i.e., axis=1):

t1 = time.time()

task2=task1.max(axis=1)

t2 = time.time()

print("Time (milliseocnds):", round((t2-t1)*1000,3))

task2

Again, the task2 is shown as an array with a shape of (50,), and it took ~6 milliseconds for this to be built (Figure 10-10).

Finally, you need to call a special computation method to evaluate the result – result = task2.compute(). The computation time is much higher here (~24 milliseconds) and you get the one-dimensional array of max values as expected (Figure 10-11). This is where all the tasks in the task graph are executed over multiple cores in a parallel fashion.

In fact, you can check the details of the task graph just by examining the dask attribute of any array such as task2 (Figure 10-12).

In the Jupyter notebook, each of these layers can be expanded to see more details. You are encouraged to check out the accompanying notebook.

DataFrames

Dask DataFrames are equally easy to use if you are already familiar with pandas. You can create a DataFrame with timeseries data using Dask’s built-in datasets module:

from dask import datasets

df = datasets.timeseries(

    start='2022-01-01',

    end='2022-01-31',

    freq='1min',

    partition_freq='1d',)

Now, if you just type df in a Jupyter notebook cell, it won’t show the data snapshot that you are used to seeing in a pandas DataFrame. This is because Dask operates on lazy evaluation and just typing df does not demand any actual computation. Instead, it will show the schema (i.e., datatypes) and the general structure information (Figure 10-13). Note that it has 30 partitions because you chose the partition_freq = '1d' in the code and the start and end dates fall on the 1^st and 31^st of the month.

If you want to see the first few entries, the familiar head method will serve that purpose and, by default, the computation will be done (i.e., the actual data will be shown) as in Figure 10-14.

Most pandas-type operations are supported. For example, to know how many unique names there are, you write the following code:

df['name'].nunique().compute()

>> 26

Now, to group by those names and compare their variances of x and y data side by side, you can write

df.groupby(by='name').var().compute()[['x','y']]

>>

         x           y

name

Alice    0.331361    0.318624

Bob      0.328595    0.336009

Charlie  0.324984    0.334246

Dan      0.329188    0.333593

Edith    0.324070    0.332390

Frank    0.340098    0.335124

<truncated output>

Direct plotting is also supported like pandas. Using a special resample method (because the data is a time series), you can plot the mean data like this (Figure 10-15):

df[['x', 'y']].resample('24h').mean().compute().plot()

Randomly accessing a partition’s data is fast but still needs to be computed to see the actual data. For example, to see all the time for the 25^th of January partition (Figure 10-16), you can write this:

df.loc['2022-01-25'].compute()

Dask Bags

Here’s a Dask Bag example that contains some JSON records. This could be randomly generated information and the code for generating such JSON data is given in the accompanying notebook/source code. You can have five JSON records (about five people) in a folder called data. You read them in a Dask Bag structure via following code (note the use of map and json.loads functions):

import dask.bag as db

import json

bag = db.read_text('data/*.json').map(json.loads)

Again, due to lazy evaluation, you cannot see inside the Bag unless you explicitly ask for that. You can use either take method for that:

bag.take(2)

This should show something like Figure 10-17. The records contain information about people’s name, occupation, phone number, and address. The record is multi-level. For example the address field has another level of data fields: address and city.

Now you can do operations like map, filter, and aggregation on this records data. For example, you may want to filter only those people whose age is over 50 and whose credit-card expiration date year is beyond 2022. You write a simple filtering function and pass it to the Bag object’s filter method. Note that you must use take or compute to get the actual computation done.

def filter_func(record):

    cond1 = record['age'] > 50

    cond2 = int(record['credit-card']['expiration-date'].split('/')[-1]) > 22

    return cond1 and cond2

bag.filter(filter_func).take(2)

This may return something like Figure 10-18.

There are many powerful usages for Dask Bags with semi-structured datasets that would have been difficult to accomplish just with an array or DataFrames.

Dask Distributed Client

All the usage examples in the earlier sections feature the formalism and lazy evaluation nature of Dask APIs (arrays, DataFrames, and Bags), but they don’t showcase the distributed/parallelized nature of computation in an obvious manner. For that, you must select and use the distributed scheduler from the Dask repertoire. It is actually a separate module or lightweight library called Dask.distributed that extends both the concurrent.futures and Dask APIs to moderate sized clusters.

The flexibility and power of the scheduler also stems from the fact that it is asynchronous and event driven. This means it can simultaneously respond to computation requests from multiple clients and track the progress of a multitude of workers that have been given tasks already. It is also capable of concurrently handling a variety of workloads coming from multiple users while also managing a dynamic worker population with possible failures and new additions.

The best thing for the user, a data scientist, is that they can use all these features and powers with pure Python code and a minimal learning curve. Cluster management or distributed scheduling is not a trivial matter to accomplish programmatically. A data scientist using Dask does not have to bother about those complexities as they are abstracted away. That’s where the theme of productive data science gets its support from libraries like Dask.

Now, if you type client in the Jupyter notebook cell, you will see a description like Figure 10-19. Note that it shows a hyperlink to a dashboard, which you will see in action soon.

If you keep expanding the Cluster Info drop-down, you may see something like Figure 10-20. Note how it shows the threads/workers of the local machine and the available system memory.

Once the scheduler is started up, it manages the distributed computing aspects by itself. However, there is a certain way to submit jobs to the scheduler using map and submit methods. Here is a (somewhat contrived) example.

Suppose you have a few datasets of random variables (generated from a specific statistical distribution) and you want to measure the differences between their max and median, and then take an average of those measurements. Each dataset may contain 1,000 values and there are 21 such datasets. Taken together, this could be a measure of some sort of outliers in the data (i.e., how much the max value is higher than the median values for a certain batch of data). You have the data generation code in the accompanying notebook. The distributions (of individual datasets) are shown in Figure 10-21.

However, more interesting things can be observed simply by looking at the dynamic dashboard that Dask provides. You can simply click on the hyperlink shown in Figure 10-20 and see something like Figure 10-22.

This is a static snapshot of the task status tab of the dashboard. When the parallel processes execute (distributed over multiple CPU cores), the graph changes and updates dynamically as all the data chunks are split and shared among workers. A good visual demonstration of this dynamic process can be seen in an article that I published at https://medium.com/productive-data-science/out-of-core-larger-than-ram-machine-learning-with-dask-9d2e5f29d733 with a hands-on example involving the Dask Machine Learning library. You are encouraged to check out this article.

There are many other tabs in this dashboard. The information tab about workers is one among them (Figure 10-23). Again, here the view is static and after the processing was finished. Therefore, you see minimal usage of memory and CPU. But for a dynamic state, those numbers will be high and constantly changing.

Dask Machine Learning Module

While Dask provides an amazing suite of parallel and out-of-core computing facilities and a straightforward set of APIs (Arrays, DataFrames, Bags, etc.,), the utility does not stop there. Going beyond the data wrangling and transformation stage, when data scientists arrive at the machine learning phase, they can still leverage Dask for doing the modeling and preprocessing tasks with the power of parallel computing. All of this can be achieved with a minimal change in their existing codebase and in pure Pythonic manner.

For ML algorithms and APIs, Dask has a separately installable module called dask-ml. Full treatment of that module is beyond the scope of this book. You are again encouraged to check out the above-mentioned article to get a feel about the API. Here, I will briefly discuss some key aspects of dask-ml.

Features and Ecosystem of Ray

Utilizing these features, a great many distributed computing tools and frameworks are being built that are powered by the engine of Ray. For an excellent reference article to get an overview of these tools, go to https://gradientflow.com/understanding-the-ray-ecosystem-and-community/. Figure 10-24 shows a visual illustration.

In this section, I will show only a couple of examples of running parallel data science workloads using Ray. You are highly encouraged to check out the official documentation (www.ray.io/docs) and try out all the great features that this library provides.

Simple Parallelization Example

Before I show the hands-on examples, I want to mention that Ray is currently built and tested for Linux and Mac OS, and the Windows version is experimental and not guaranteed to be stable. Therefore, you are encouraged to practice Ray examples in a Linux environment or create a virtual machine (VM) on your Windows platform, install Ray, and continue.

For example, the following examples are run inside an Ubuntu Linux 20.04 environment that runs within a VM managed by Oracle Virtual Box software (installed on a Windows 11 laptop). The VM has also been assigned four logical CPU cores by the creator/user (Figure 10-25). This is important to note as the default starting number for the CPU cores may be only one and that will not demonstrate the expected speed-up for parallel processing tasks. A detailed guide on how to create such a VM is given in this article (https://brb.nci.nih.gov/seqtools/installUbuntu.html). If you are working on native Linux or Mac OS machine, then this step is unnecessary.

If you click this hyperlink, you will see the Ray dashboard with workers and their status, as shown in Figure 10-26 (quite like the Dask dashboard discussed earlier).

Note how you call the build_stats_ray function with a .remote() method and how you wrap that with the ray.get() method to run everything in parallel. The takeaway is that although Ray offers a great many features, you must learn how to properly take advantage of them and how to submit a parallelizable task to the Ray cluster by pipelining the sub-components in correct order. Figure 10-27 shows the idea.

Ray Dataset for Distributed Loading and Compute

Ray Datasets (https://docs.ray.io/en/latest/data/dataset.html) are the standard (and recommended) way to load and exchange data in the Ray ecosystem. These objects provide basic distributed data transformations such as map, filter, and repartition, and play well with a wide variety of file formats, data sources, and distributed frameworks for easy loading and conversion. They are also specifically designed to load and preprocess data with high performance for distributed ML training pipelines built with Ray such as Ray-Train. (https://docs.ray.io/en/latest/train/train.html#train-docs).

Ray Datasets are a good candidate for the last-mile data processing blocks (before data is fed into a parallelized ML task flow) where the initial data sources are traditional RDBMS, output of ETL pipeline, or even Spark DataFrames.

Previously, I talked about Apache Arrow and how these modern data storage formats are revolutionizing the data science world. Ray Datasets, at their core, implement distributed Arrow. Each Dataset is essentially a list of Ray object references to blocks that hold Arrow tables (or Python lists in some cases). The presence of such block-level structure allows the parallelism and compatibility with distributed ML training. In this manner, Ray Datasets are similar to what you saw with Dask. Moreover, since Datasets are just lists of Ray object references, they can be freely (almost no memory operation overhead) exchanged between Ray tasks, actors, and libraries. This lets you have tremendous flexibility with their usage and integration, and it improves the system performance.

As mentioned, Ray Datasets work with almost every kind of data sources that you use in your everyday work. Figure 10-28 shows a partial snapshot of their input compatibility.

So, by default, it has created 200 blocks of object reference and also assigned a schema of integer to the data. This parallelism and data type integration inherently makes the Dataset more efficient than traditional data sources like pandas DataFrame.

The result looks like Figure 10-29.