What Dask is for ?

• Problem: Python is powerful and user friendly but it doesn’t scale well
• Solution: Dask enables to scale Python natively

What is Dask ?

Python library for parallel and distributed computing

1. Scales Numpy, Pandas and Scikit-Learn
2. General purpose computing/parallelization framework

Why use Dask ?

• Allow to process data than is larger than available memory for a single machine
• Parallel execution for faster processing
• Distribute computation for large datasets

Dask provides several APIs

• Dataframes
• Arrays
• Bags
• Delayed
• Futures

Dataframes

import pandas as pd

df = pd.read_csv("file.csv")
result = df.groupby(df.name).amount.mean()

import dask.dataframe as dd

df = dd.read_csv("file.csv")
result = df.groupby(df.name).amount.mean()

result = result.compute()  # Compute to get pandas result

Arrays

import numpy as np

x = np.random.random((10000, 10000))
y = (x + x.T) - x.mean(axis=1)

import dask.array as da

x = da.random.random((10000, 10000))
y = (x + x.T) - x.mean(axis=1)

Bags

import dask.bag as db

# Read large datasets in parallel
lines = db.read_text("s3://mybucket/data.*.json")
records = (lines
    .map(json.loads)
    .filter(lambda d: d["value"] > 0)
)
df = records.to_dask_dataframe()

Delayed

from dask.distributed import LocalCluster
client = LocalCluster().get_client()

# Submit work to happen in parallel
results = []
for filename in filenames:
    data = client.submit(load, filename)
    result = client.submit(process, data)
    results.append(result)

# Gather results back to local computer
results = client.gather(results)

Futures

from dask.distributed import LocalCluster
client = LocalCluster().get_client()
futures = client.map(score, x_values)

best = -1
for future in as_completed(futures):
   y = future.result()
   if y > best:
       best = y

Two levels of API

First, produce a task graph

High level collections are used to generate task graphs

First, produce a task graph

Create an array of ones

import dask.array as da

x = da.ones(15, chunks=(5,))

First, produce a task graph

import dask.array as da

x = da.ones(15, chunks=(5,))
y = sum()

First, produce a task graph

Create an 2d-array of ones and sum it

import dask.array as da

x = da.ones((15,15), chunks=(5,5))
y = x.sum()

First, produce a task graph

Add array to its transpose

import dask.array as da

x = da.ones((15,15), chunks=(5,5))
y = x +x.T

First, produce a task graph

Matrix multiplication

import dask.array as da

x = da.ones((15,15), chunks=(5,5))
y = da.ones((15,15), chunks=(5,5))
r = da.matmul(x,y)

Dask graph

import dask.array as da

x = da.ones((15,15), chunks=(5,5))
y = da.ones((15,15), chunks=(5,5))
r = da.matmul(x,y)

x

Dask graph

• Every operations/tasks submit to Dask are turned to a graph
• Dask is lazily evaluated
• The real computation is performed by executing the graph

Then, compute the calculation

Use compute() to execute the graph and get the result

Then, compute the calculation

Dask execution

• Task graphs can be executed by schedulers on a single machine or a cluster
• Dask offers several backend execution systems, resilience to failures

Dask execution

Dask execution

• Client: interacts with the Dask cluster, submits task
• Scheduler: is in charge of executing the Dask graph, sends task to the workers
• Workers: compute tasks as directed by the scheduler, store and serve computed results to other workers or clients

Local execution

• Deploy Dask cluster on a single machine
• Configure to use threads or multiprocessing

from dask.distributed import LocalCluster

cluster = LocalCluster()
client = cluster.get_client()

Distributed execution

• Deploy Dask cluster on a distributed hardware
• Dask can work with:
  • popular HPC job submission systems like SLURM, PBS, SGE, LSF, Torque, Condor
  • Kubernetes

from dask_kubernetes.operator import KubeCluster
cluster = KubeCluster(
   name="my-dask-cluster",
   image="ghcr.io/dask/dask:latest",
   resources={"requests": {"memory": "2Gi"}, "limits": {"memory": "64Gi"}},
)
cluster.scale(10)
client = cluster.get_client()

Use Dashboard

Scikit-learn/Joblib

model = Model(...,n_jobs=n)
model.fit(X,y)

Scikit-learn/Joblib/Dask

from joblib import parallel_backend
model = Model(...,n_jobs=n)
with parallel_backend("dask"):
    model.fit(X,y)

Dask-ML

• Provides scalable machine learning alongside popular machine learning libraries
• Work with
  • Scikit-Learn,
  • XGBoost
  • PyTorch
  • Tensorflow/Keras

via pip

python -m pip install "dask[complete]"    # Install everything

Extras package

• Use Dask on queuing systems like PBS, Slurm, MOAB, SGE, LSF, and HTCondor.

pip install dask-jobqueue

• Use Dask on Kubernetes

pip install dask-kubernetes

• Use Dask with machine learning framework

pip install dask-ml

Dask Tutorial

Dask tutorial

Try to follow by order of importance:

• Dask Dataframes
• Distributed
• Delayed
• Parallel and Distributed Machine Learning
• Next, if you have more time
  • Array
  • Futures

Pangeo tutorial or finish deploying your computing platform

Just try some use cases here

or

Finish yesterday deployment (needed for tomorrow).

Dask presentations

Just use the Dask slidedeck