Real-time collaboration for Jupyter Notebooks, Linux Terminals, LaTeX, VS Code, R IDE, and more,
all in one place. Commercial Alternative to JupyterHub.

Views: ¹⁹⁴⁴

Kernel: Python 3 (system-wide)

Dask in Python 3 (Ubuntu Linux)

https://docs.dask.org/en/latest/

In [1]:

import dask
dask.__version__

Out[1]:

'2.21.0'

In [2]:

import distributed
distributed.__version__

Out[2]:

'2.21.0'

In [3]:

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

Out[3]:

<dask.config.set at 0x7f29fc3ea2e8>

In [4]:

dask.config.config

Out[4]:

{'temporary-directory': '/home/user/tmp',
 'dataframe': {'shuffle-compression': None},
 'array': {'svg': {'size': 120}},
 'optimization': {'fuse': {'active': True,
   'ave-width': 1,
   'max-width': None,
   'max-height': inf,
   'max-depth-new-edges': None,
   'subgraphs': None,
   'rename-keys': True}},
 'distributed': {'version': 2,
  'scheduler': {'allowed-failures': 3,
   'bandwidth': 100000000,
   'blocked-handlers': [],
   'default-data-size': '1kiB',
   'events-cleanup-delay': '1h',
   'idle-timeout': None,
   'transition-log-length': 100000,
   'work-stealing': True,
   'work-stealing-interval': '100ms',
   'worker-ttl': None,
   'pickle': True,
   'preload': [],
   'preload-argv': [],
   'unknown-task-duration': '500ms',
   'default-task-durations': {'rechunk-split': '1us', 'shuffle-split': '1us'},
   'validate': False,
   'dashboard': {'status': {'task-stream-length': 1000},
    'tasks': {'task-stream-length': 100000},
    'tls': {'ca-file': None, 'key': None, 'cert': None},
    'bokeh-application': {'allow_websocket_origin': ['*'],
     'keep_alive_milliseconds': 500,
     'check_unused_sessions_milliseconds': 500}},
   'locks': {'lease-validation-interval': '10s', 'lease-timeout': '30s'},
   'http': {'routes': ['distributed.http.scheduler.prometheus',
     'distributed.http.scheduler.info',
     'distributed.http.scheduler.json',
     'distributed.http.health',
     'distributed.http.proxy',
     'distributed.http.statics']}},
  'worker': {'blocked-handlers': [],
   'multiprocessing-method': 'spawn',
   'use-file-locking': True,
   'connections': {'outgoing': 50, 'incoming': 10},
   'preload': [],
   'preload-argv': [],
   'daemon': True,
   'validate': False,
   'lifetime': {'duration': None, 'stagger': '0 seconds', 'restart': False},
   'profile': {'interval': '10ms', 'cycle': '1000ms', 'low-level': False},
   'memory': {'target': 0.6, 'spill': 0.7, 'pause': 0.8, 'terminate': 0.95},
   'http': {'routes': ['distributed.http.worker.prometheus',
     'distributed.http.health',
     'distributed.http.statics']}},
  'nanny': {'preload': [], 'preload-argv': []},
  'client': {'heartbeat': '5s', 'scheduler-info-interval': '2s'},
  'deploy': {'lost-worker-timeout': '15s', 'cluster-repr-interval': '500ms'},
  'adaptive': {'interval': '1s',
   'target-duration': '5s',
   'minimum': 0,
   'maximum': inf,
   'wait-count': 3},
  'comm': {'retry': {'count': 0, 'delay': {'min': '1s', 'max': '20s'}},
   'compression': 'auto',
   'offload': '10MiB',
   'default-scheme': 'tcp',
   'socket-backlog': 2048,
   'recent-messages-log-length': 0,
   'zstd': {'level': 3, 'threads': 0},
   'timeouts': {'connect': '10s', 'tcp': '30s'},
   'require-encryption': None,
   'tls': {'ciphers': None,
    'ca-file': None,
    'scheduler': {'cert': None, 'key': None},
    'worker': {'key': None, 'cert': None},
    'client': {'key': None, 'cert': None}}},
  'dashboard': {'link': '{scheme}://{host}:{port}/status',
   'export-tool': False,
   'graph-max-items': 5000},
  'admin': {'tick': {'interval': '20ms', 'limit': '3s'},
   'max-error-length': 10000,
   'log-length': 10000,
   'log-format': '%(name)s - %(levelname)s - %(message)s',
   'pdb-on-err': False}},
 'rmm': {'pool-size': None},
 'ucx': {'tcp': None,
  'nvlink': None,
  'infiniband': None,
  'rdmacm': None,
  'cuda_copy': None,
  'net-devices': None,
  'reuse-endpoints': True},
 'scheduler': {'work-stealing': True}}

In [6]:

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

Out[6]:

Start dash-scheduler and dash-worker with --dashboard-prefix b9bacd7b-6cee-402c-88ed-9d74b07f29a1/port/8787

The dashboard is actually at https://cocalc.com/{{ THE PROJECT UUID }}/port/8787/status

Websocket forwarding doesn't work, though ... hmm...

alternatively, start an X11 desktop in cocalc and run google-chrome at http://127.0.0.1:8787/status

data array similar to numpy arrays

In [7]:

import dask.array as da

In [8]:

client.restart()

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

Out[8]:

In [9]:

x = x.rechunk({0: 500, 1: 500}).persist()
x

Out[9]:

In [10]:

client.scatter(x)

Out[10]:

In [11]:

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

Out[11]:

In [12]:

z.shape

Out[12]:

(20,)

In [12]:

out = z.compute()
out

Out[12]:

array([0.99706421, 1.03277113, 0.9993643 , 0.99231638, 1.02139168,
       0.98631926, 0.99280159, 0.97743904, 0.99200793, 1.00457694,
       1.01779522, 0.98179355, 1.01977627, 1.01330775, 1.00401255,
       0.98929948, 0.98495306, 0.99648525, 0.98166991, 1.01806776])

In [13]:

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

Out[13]:

0.9974348104666977

In [14]:

del x, y, z, rz, out

sum 1000 ints

In [12]:

client.restart()

import dask.bag as db
nums = db.from_sequence(range(1234), npartitions=10)
nums.sum().compute()

Out[12]:

760761

functions and native lists

In [17]:

import numpy as np
xx = dask.array.array(np.arange(100))
xx = xx.reshape(10, -1)
xx = xx.rechunk(2, 2)
xx = client.persist(xx)
xx

Out[17]:

In [18]:

y = client.submit(lambda x : x.dot(x.T), xx)
y.result()

Out[18]:

In [18]:

client.restart()

import dask.array as da
from time import sleep
from random import random
import numpy as np

delay = lambda : sleep(.2 * random())

def formula(x):
    delay()
    return x**3 - x**2 + 1


def neg(a):
    delay()
    return -a

def dup(a):
    delay()
    return 2 * a

def mysum(a, b):
    delay()
    return lambda a, b: a + b

N = 100
A = client.map(formula, range(N))
B = client.map(neg, A[:N // 3])
B.extend(client.map(dup, A[N // 3:2 * N // 3]))
B.extend(B[2*N//3:])
C = client.map(mysum, A, B)
total = client.submit(sum, B)
total.result()

Out[18]:

8212721

loops?

In [19]:

from time import sleep
from random import random

@dask.delayed
def fib(x):
    sleep(random())
    if x <= 1:
        return 1
    else:
        a, b = fib(x - 2), fib(x - 1)
        a, b = dask.compute(a, b)
        return a + b

In [20]:

dask.compute((fib(i) for i in range(5)), 10)

Out[20]:

([1, 1, 2, 3, 5], 10)

Dask Bags

In [21]:

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

Out[21]:

dask.bag<from_sequence, npartitions=50>

In [22]:

import operator

In [23]:

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()

Out[23]:

[(False, -16256), (True, -16240)]

Dask Delayed

In [14]:

from dask import delayed

In [15]:

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

In [16]:

from time import sleep
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()

Out[16]:

30

In [17]:

z.vizualize(filename=os.path.expanduser('~/dask-delayed-1.png'), format='png')

Out[17]:

Delayed('vizualize-ad3fba49-bb55-4757-a0ce-668e7fb3aac8')

In [18]:

z.vizualize()

Out[18]:

Delayed('vizualize-3dde23e6-bc93-4e01-a407-6b5d33a0ddbe')

Dask ML

https://dask-ml.readthedocs.io/en/latest/

In [0]:

import dask_ml
dask_ml.__version__

In [0]:

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

In [0]:

lr = LogisticRegression()
lr

In [0]:

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

In [0]:

import joblib
joblib.__version__

In [0]:

In [34]:

client.restart()

Out[34]:

In [0]:

Ad-hoc Local Cluster

In [34]:

from dask.distributed import Client, LocalCluster
import dask.array as da

cluster = LocalCluster(
    n_workers=3,
    threads_per_worker=1,
    processes=True,
    diagnostics_port=None,
)
client = Client(cluster)
x = da.random.random((300, 300), chunks=(10, 10))
y = x + x.T
z = (y.mean(axis=1) / y.shape[0]).std()
print(z.compute())

Out[34]:

/usr/local/lib/python3.6/dist-packages/distributed/dashboard/core.py:79: UserWarning: 
Port 8787 is already in use. 
Perhaps you already have a cluster running?
Hosting the diagnostics dashboard on a random port instead.
  warnings.warn("\n" + msg)

7.34683839787855e-05

In [0]:

In [0]:

In [0]:

Real-time collaboration for Jupyter Notebooks, Linux Terminals, LaTeX, VS Code, R IDE, and more,
all in one place. Commercial Alternative to JupyterHub.

Dask in Python 3 (Ubuntu Linux)

data array similar to numpy arrays

sum 1000 ints

functions and native lists

loops?

Dask Bags

Dask Delayed

Dask ML

Ad-hoc Local Cluster

Product

Resources

Company

Real-time collaboration for Jupyter Notebooks, Linux Terminals, LaTeX, VS Code, R IDE, and more, all in one place. Commercial Alternative to JupyterHub.

Dask in Python 3 (Ubuntu Linux)

data array similar to numpy arrays

sum 1000 ints

functions and native lists

loops?

Dask Bags

Dask Delayed

Dask ML

Ad-hoc Local Cluster

Real-time collaboration for Jupyter Notebooks, Linux Terminals, LaTeX, VS Code, R IDE, and more,
all in one place. Commercial Alternative to JupyterHub.