dask.delayed - Parallel Computing in Python¶

Table of Contents¶

• Introduction
  • Directed Acyclic Graph (DAG) of Execution
• Common Way of Using dask.delayed API
• Example 1
• Example 2
• Example 3
• Example 4
• Example 5
• References

Introduction ¶

Dask is a python library that provides a list of APIs for performing the computation in parallel on a single computer as well as a cluster of computers. We have already created tutorial on dask.bag API which is spark like API for parallelizing big data. Dask has other APIs like dask.delayed, dask.dataframe, dask.array and dask.distributed whose introduction we have covered in the previous tutorial. We suggest that you go through that tutorial to get basic information about the dask. We'll be specifically concentrating on dask.delayed API as a part of this tutorial.

The dask.delayed provides a very flexible API which lets us parallelize our python functions. It's very suitable for problems when it doesn't involve data structures like dask.array or dask.dataframe. The API of dask.delayed is simplified which lets us parallelize our python code very easily.

Directed Acyclic Graph (DAG) of Execution ¶

The dask.delayed API is used to convert normal function to lazy function. When a function is converted from normal to lazy, it prevents function to execute immediately. Instead, its execution is delayed in the future. Dask can easily run these lazy functions in parallel.

The dask.delayed API keeps on creating a directed acyclic graph of these lazy functions in the background. This graph keeps track of the execution sequence of these lazy functions which helps it understand which parts can be executed in parallel. When we evaluate this graph by calling compute() method of dask.delayed API then it evaluates all functions by running them in parallel. We'll be visualizing graphs created by dask when explaining the usage of dask.delayed API.

This ends our small introduction of dask.delayed API. We'll be explaining its usage with few examples below.

So without further delay, let’s get started with the coding part.

We'll first import all the necessary libraries for a tutorial.

Common Way of Using dask.delayed API ¶

We can easily parallelize our normal python code using dask by following list of below steps:

• Find out the list of time-consuming functions in code that are getting executed more than once.
• Convert a list of time-consuming functions from normal to lazy by using dask.delayed. All functions calling these lazy functions will also become lazy.
• Execute last lazy function. There is generally two scenarios of this.
  • Scenario 1: You only need to execute one final function which calls a list of lazy functions. If this is the scenario then you can call compute() on final lazy function and it'll run all lazy functions it depends on in parallel.
  • Scenario 2: If there are more than one final lazy functions then we can use compute() method of dask. We need to pass all lazy functions to this method and it'll run all of them in parallel and returns results for all of them.

We'll be explaining these steps below with few examples to make things clear. We'll be creating normal python functions which takes time to execute normally but runs faster when parallelized using dask.

Optional Step: Start Dask Client to Analyze Results¶

Dask provides an optional dashboard which we can utilize to see the performance of operations getting performed in parallel. This step is optional if you don't want to analyze results but it can be very useful to debug things.

We suggest that you pass n_workers same as a number of cores on your PC and threads_per_worker as 1 if you are running code on local.

You can open the URL mentioned next to the dashboard. It'll have a list of tabs that provides information like running time of the task, CPU usage, memory usage, directed graphs, etc. All this can be useful for analyzing parallel execution of tasks. You'll be able to see the whole directed acyclic graph of lazy functions as well in the Graph section of the dashboard. They will be all marked blue once that particular lazy function node has completed execution. If your function has many lazy functions then you'll be able to see changing their values in real-time to blue when they complete.

Example 1 ¶

Below we have created a simple python example which takes the square of two numbers and adds them. We have created a function for getting square of numbers. We have artificially added sleep time of 1 second to function to mimic a real-life situations where functions can take time to execute. We are calling this function 2 times to get a square of 2 different numbers before adding them. As per below pure python code, it executes square function sequentially which results in taking 2 seconds for code to complete.

We'll now try to convert the above example to the dask version so that it can run fast using parallelization. From the above code, we can easily notice that square function takes time to execute. We can even notice that each execution of the square is totally independent of any other execution of the same function which makes it ideal for parallel execution.

We'll convert each square function execution to lazy function by calling delayed on them as explained below.

We can notice from above cell execution that it executed immediately. It's because of dask.delayed API has just converted normal square execution to lazy execution and has created an execution graph as of now. But it has not actually executed functions. We can see that the return type of res is Delayed object instead of the result.

We can make dask execute code in parallel by calling compute() method on the final res object. It'll evaluate the graph and execute a square function in parallel.

We can notice from the above cell that it takes less time to compute the sum of squares using dask.delayed API.

We can even visualize DAG created by dask by calling visualize() method on the final res object.

Example 2 ¶

Below we are explaining another usage of dask.delayed API. Here we are creating an array of random numbers of size 10x1000. We then want to compute the correlation between the first row of the array and all other rows. We have implemented this solution as a loop. We also have designed the function corr which calculates correlation. We have added a time delay of 1 second into function execution to mimic real-life situations.

We can notice that the above code takes 10 seconds for execution because there are 10 rows in an array. We are computing a correlation between the first row and all other rows, all of which take 1 second.

We are below converting the above code to dask delayed code so that it runs in parallel. We have used delayed at two places this time. First on corr() function and then on np.mean() functions.

Like the previous example, we can notice that the above code with lazy functions executes fast because actual execution of code has not happened yet.

We'll not call compute() on final avg_corr variable to complete execution.

We can notice that it completes execution is quite less time compared to normal sequential execution.

Below we are visualizing the execution graph of the above code.

Example 3 ¶

Below we are demonstrating usage of dask.delayed API using the third example. We are first looping through number 1-10 taking the square of the number if the number is divisible by 2 else taking power 3 of number. We have defined function for taking a square and power 3 of numbers. We have also added time delay in function so that it looks like a real-life situation.

We can see above that when we run above loop in parallel then it takes 10 seconds for it to complete. We can easily convert this code using dask.delayed as explained below. This example demonstrates how conditional functions can also be parallelized using dask.

We have made use of @delayed annotation this time instead of wrapping calls of function like previous examples. We can also use the annotation above function rather than wrapping each call of function into delayed.

Example 4 ¶

As a part of our 4th example, we'll be working with some files. We are creating 30 CSV files for 30 days of temperature data where the temperature is recorded every 5 seconds. We are first creating CSV files into a separate folder for each day. Each CSV file will have 17280 temperature values for that day recorded at every 5 seconds. We are filling temperature data with a random temperature between 1-50.

Below we are looping through temperature file of each day, reading it, taking an average of that day, and adding it to list. We are then taking an average of all days combined. Our main aim of this example was to explain the usage of dask.delayed API working with files.

We can see that it takes nearly 12 seconds for the above code to complete.

We have converted it to dask code below which completes in nearly 3 seconds.

Example 5 ¶

Our fifth and last example is designed to explain the usage of compute() method available directly from dask which takes an input list of lazy objects and can evaluate them.

Below we are looping through 1-10, creating an array of random numbers of size 100 between 1-100. We are then taking sum and count of numbers divisible by 2. We add all sums and counts to an array. We then sum the array of sums and divide it by sum of counts array to get average. We have added time delay to function so that it takes time to mimicking real-life situations.

Below we have converted the above example to dask.delayed version. We have annotated both functions with @delayed annotation. We have evaluated both totals_orig and counts_orig array which are list of delayed objects using compute() function. We can see that this one takes nearly 1 second to complete compared to the previous version.

This ends our small tutorial explaining the usage of dask.delayed API for parallelizing python code. Please feel free to let us know your views in the comments section.

References ¶

• dask.bag - Parallel Computing in Python
• concurrent.futures - Multiprocessing and Multithreading API of Python
• Numba - Make Your Python Functions Run Like C/C++