<jupyter_start><jupyter_text>Dask delayed and persist experimentation<jupyter_code># Useful info here: https://www.youtube.com/watch?v=SHqFmynRxVU

import time
import random

# The functions below could be thought of as:
# inc: load data, dec: process data, add: combine data

def inc(x):
    time.sleep(random.random())
    return x + 1

def dec(x):
    time.sleep(random.random())
    return x - 1

def add(x, y):
    time.sleep(random.random())
    return x + y

# Running sequentially
start = time.time()
x = inc(1)
y = dec(2)
z = add(x, y)
end = time.time()
print(f'z: {z}, time elapsed: {round(end - start, 2)}')


# Can use dask to parallelize - inc and dec can be run at same time
# Specifically, we can use dask delayed as a decorator, making our functions lazy by default
from dask import delayed

@delayed
def inc(x):
    time.sleep(random.random())
    return x + 1

@delayed
def dec(x):
    time.sleep(random.random())
    return x - 1

@delayed
def add(x, y):
    time.sleep(random.random())
    return x + y

# Running code now will return instantly, giving us a dask delayed object (i.e. z)
start = time.time()
x = inc(1)
y = dec(2)
z = add(x, y)
end = time.time()
print(f'z: {z}, time elapsed: {round(end - start, 2)}')
z.visualize()
start = time.time()
z.compute()
end = time.time()
print(f'z: {z}, time elapsed: {round(end - start, 2)}')<jupyter_output>z: Delayed('add-eb0e4523-b063-41ba-9ad8-7e54569d0e25'), time elapsed: 1.32<jupyter_text>Experiment with dask delayed and persist<jupyter_code>import dask
from dask import persist
from distributed import Client
client = Client()<jupyter_output><empty_output><jupyter_text>No call to `persist()`<jupyter_code>%%time 
zs = []
for i in range(128):
    x = inc(i)
    y = dec(x)
    z = add(x, y)
    zs.append(z)
L = zs 
while len(L) > 1:
    new_L = []
    for i in range(0, len(L), 2):
        lazy = add(L[i], L[i + 1])
        new_L.append(lazy)
    L = new_L
    
len(L)
%%time
L = dask.compute(L)
len(L)<jupyter_output><empty_output><jupyter_text>Call to `persist()`<jupyter_code>%%time 
zs = []
for i in range(128):
    x = inc(i)
    y = dec(x)
    z = add(x, y)
    zs.append(z)

zs = persist(zs)

L = zs 
while len(L) > 1:
    new_L = []
    for i in range(0, len(L), 2):
        lazy = add(L[i], L[i + 1])
        new_L.append(lazy)
    L = new_L
    
len(L)
%%time
L  = dask.compute(L)
len(L)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.152789342214821, 'fraction_numerical': 0.020399666944213156, 'mean_word_length': 3.4832089552238807, 'pattern_counts': {'":': 0, '<': 16, '<?xml version=': 0, '>': 18, 'https://': 1, 'lorem ipsum': 0, 'www.': 1, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '4', 'dir': 'jupyter-structured-clean-dedup', 'id': '523755', 'license': 'MIT', 'n_tokens_mistral': 1003, 'n_tokens_neox': 962, 'n_words': 270, 'path': 'notebooks/computer_science/dask-overview/dask-delayed-and-persist.ipynb', 'repo_name': 'NathanielDake/NathanielDake.github.io', 'size': '2402'}