Dask in Python 3 (Ubuntu Linux)

import dask
dask.__version__

import dask
import dask.distributed
import os
dask.config.set({
    'temporary_directory': os.path.expanduser('~/tmp'),
    'scheduler.work-stealing': True
})

dask.config.config

from dask.distributed import Client
client = Client('127.0.0.1:8786')
client

import dask.array as da

x = da.random.random((3000, 3000), chunks=(300, 300))
x

y = x + x.T
z = y[::50, 1000:].mean(axis=1)
z

z.shape

out = z.compute()
out

(z[:100].sum() / z[:100].shape[0]).compute()

fn = lambda x : x**3 - x**2 + 1

def neg(a):
    import time
    time.sleep(.1)
    return -a

A = client.map(fn, range(10))
B = client.map(neg, A)
C = client.map(lambda a, b: a+b, A, B)
total = client.submit(sum, B)
total.result()

client.gather(C)

def fib(x):
    if x <= 1:
        return 1
    else:
        return fib(x - 2) + fib(x - 1)

[fib(_) for _ in range(10)]

import dask.bag as db
b1 = db.from_sequence(range(-1000, 1000), npartitions=50)
b1

import operator

is_odd = lambda x : x % 2 == 0

#b1.groupby(is_odd).map(lambda k_v : (k_v[0], sum(k_v[1]))).compute()
b1.foldby(is_odd, operator.add, 0).compute()

from dask import delayed

inc = lambda x : x+1
from operator import add

z = delayed(0)
for i in range(5):
    x = delayed(inc)(i)
    y = delayed(inc)(delayed(add)(i, x))
    z = delayed(add)(z, y)
z.compute()

z.vizualize(filename='dask-delayed-1.svg')

import dask_ml

from dask_ml.preprocessing import Categorizer, DummyEncoder
from dask_ml.linear_model import LogisticRegression

lr = LogisticRegression()
lr

from sklearn.externals.joblib import parallel_backend
with parallel_backend('dask') as pb:
    print(pb[0])