<jupyter_start><jupyter_text><img src="http://dask.readthedocs.io/en/latest/_images/dask_horizontal.svg"      width="30%"      align=right     alt="Dask logo">Custom Workloads-------------------------*Because not all problems are dataframes*<jupyter_code>from dask_kubernetes import KubeCluster
cluster = KubeCluster(n_workers=10)
cluster
from dask.distributed import Client, progress
c = Client(cluster)
c
from time import sleep

def inc(x):
    from random import random
    sleep(random())
    return x + 1

def double(x):
    from random import random
    sleep(random())
    return 2 * x
    
def add(x, y):
    from random import random
    sleep(random())
    return x + y 
inc(1)
future = c.submit(inc, 1)  # returns immediately with pending future
future
future  # scheduler and client talk constantly
future.result()<jupyter_output><empty_output><jupyter_text>Submit many tasksWe submit many tasks that depend on each other in a normal Python for loop<jupyter_code>%%time
zs = []
for i in range(256):
    x = c.submit(inc, i)     # x = inc(i)
    y = c.submit(double, x)  # y = inc(x)
    z = c.submit(add, x, y)  # z = inc(y)
    zs.append(z)
    
total = c.submit(sum, zs)
total.result()<jupyter_output><empty_output><jupyter_text>Custom computation: Tree summationAs an example of a non-trivial algorithm, consider the classic tree reduction.  We accomplish this with a nested for loop and a bit of normal Python logic.```finish           total             single output    ^          /        \    |        c1          c2        neighbors merge    |       /  \        /  \    |     b1    b2    b3    b4     neighbors merge    ^    / \   / \   / \   / \start   a1 a2 a3 a4 a5 a6 a7 a8    many inputs```<jupyter_code>L = zs
while len(L) > 1:
    new_L = []
    for i in range(0, len(L), 2):
        future = c.submit(add, L[i], L[i + 1])  # add neighbors
        new_L.append(future)
    L = new_L                                   # swap old list for new
   
del L<jupyter_output><empty_output><jupyter_text>Visualize Computation<jupyter_code>from dask import delayed, visualize

L = zs
while len(L) > 1:
    new_L = []
    for i in range(0, len(L), 2):
        future = delayed(add)(L[i], L[i + 1])  # add neighbors
        new_L.append(future)
    L = new_L                                   # swap old list for new
    
visualize(*L)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.11559478772593526, 'fraction_numerical': 0.013871374527112233, 'mean_word_length': 2.4097421203438394, 'pattern_counts': {'":': 0, '<': 18, '<?xml version=': 0, '>': 20, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '4', 'dir': 'jupyter-structured-clean-dedup', 'id': '44817', 'license': 'MIT', 'n_tokens_mistral': 806, 'n_tokens_neox': 783, 'n_words': 233, 'path': 'deployments/yellowstone.pangeo.io/image/custom-computations.ipynb', 'repo_name': 'andersy005/yellowstone-deploy', 'size': '2379'}