Slides

In the beginning...

Scientific Python goes back to 1995

Numpy (2005, roots back to 1995)
Scipy (2000, roots back to 1998, wrapped code even older)
IPython (2001)
Matplotlib (2002)
Cython (2007, roots back to 2002)
Pandas (2009)

Scale Up	Scale Out
- C, C++, Fortran, Cython, ...	- MPI
- CUDA-C, OpenCL	- Spark, MapReduce, ...

Blocked Algorithms

Blocked Mean

x = h5py.File('data.hdf5')['x']    # Trillion element array on disk

sums = []
counts = []
N = 1000000
for i in range(N):                  # One million times
    chunk = x[N*i: N*(i+1)]         # Pull out chunk
    sums.append(np.sum(chunk))      # Sum chunk
    counts.append(len(chunk))       # Count chunk

result = sum(sums) / sum(counts)    # Aggregate results

Blocked Algorithms

x.mean()

x + x.T

x.dot(x.T + 1)

x.dot(x.T + 1) - x.mean()

(x.dot(x.T + 1) - x.mean()).std()

U, S, V = da.linalg.svd(x + x.T - x.mean())

Supports most of the Pandas api

# Reductions
df.passenger_count.mean()

# Groupby operations
df.groupby(df.passenger_count).trip_distance.mean()

# Timeseries operations
df.trip_distance.resample('h').mean()

# many many more...

Dask Delayed

Tool for creating arbitrary task graphs
Dead simple interface (one function)
Plays well with existing code (with some caveats)

# Wrap functions to make them lazy
delayed(function)(*args, **kwargs) -> Delayed

# Wrap data to make attribute access lazy
delayed(data) -> Delayed

Dask Delayed

results = {}

for a in A:
    for b in B:
        for c in C:
            r1 = fit1(x, a)
            r2 = fit2(r1, b)
            r3 = fit3(r2, c)
            results[a, b, c] = r3
best = score(results)
_

Dask Delayed

from dask import delayed, compute

results = {}

for a in A:
    for b in B:
        for c in C:
            r1 = delayed(fit1)(x, a)
            r2 = delayed(fit2)(r1, b)
            r3 = delayed(fit3)(r2, c)
            results[a, b, c] = r3
best = delayed(score)(results)
result = best.compute()

Easy to get started

conda/pip installable

$ conda install dask distributed -c conda-forge
$ pip install dask[complete] distributed --upgrade

$ conda install numba

Easy to get started

>>> from dask.distributed import Client
>>> e = Client()  # sets up local cluster

$ dask-scheduler

$ dask-worker scheduler-hostname:8786
$ dask-worker scheduler-hostname:8786

Scaling Numerical Python with

Dask & Numba