<jupyter_start><jupyter_text>Dask graph computationHere we illustrate Dask graph for computation, the lazy mode, and parrallelism with distributed<jupyter_code>import dask.array as da<jupyter_output><empty_output><jupyter_text>Graph for chunk of operation<jupyter_code>x = da.ones((6, 1), chunks=(2, 1))
x.visualize()
y = 2 + da.ones((6, 1), chunks=(2,1))
y.visualize()
z = x.T+y
z.visualize()
x = da.triu(da.ones((3, 3), chunks=(1, 1)))
x.visualize()
print("Compare this print of x:")
print(x)
print("\nWith this one:")
print(x.compute())
x = da.triu(da.ones((4, 4), chunks=(2, 2)))
x.visualize()<jupyter_output><empty_output><jupyter_text>Graph for lazy operationsSee here https://github.com/dask/dask-examples/blob/master/delayed.ipynb for more<jupyter_code># Init dummy functions
import time
import random

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

def dec(x):
    time.sleep(3)
    return x - 1
    
def add(x, y):
    time.sleep(1)
    return x + y
%%time
# Bench these sequential list of function calls:
x = inc(1) # Return 1 + 1 = 2
y = dec(2) # Return 2 - 1 = 1
z = add(x, y) # Return 2 + 1 = 3
print("z =",z,"\n")<jupyter_output>z = 3 

CPU times: user 4 ms, sys: 5 ms, total: 9 ms
Wall time: 6.01 s<jupyter_text>note that the x and y computations are performed sequentially, hence the 6sec of wall time<jupyter_code># Make these functions to be "lazy", so that they are executed only when needed:
import dask
inc = dask.delayed(inc)
dec = dask.delayed(dec)
add = dask.delayed(add)
%%time
# Bench the same sequential list of function calls, but let Dask decide if they should be computed or not ("lazy" mode):
x = inc(1)
y = dec(2)
z = add(x, y)
print("z =",z,"\n")<jupyter_output>z = Delayed('add-8daaf9e5-0c24-4612-bee6-ed94e53747d9') 

CPU times: user 1 ms, sys: 0 ns, total: 1 ms
Wall time: 763 µs<jupyter_text>see:- that `z` is now a `delayed` Dask object, the result (3) is not returned- how the wall time is much shorter, that's because the computation was not performed, only the graph was built<jupyter_code># This is the graph:
z.visualize(rankdir='LR')
%%time
# In order to realy get the computation done, simple call "compute":
print("z =",z.compute(),"\n")<jupyter_output>z = 3 

CPU times: user 8 ms, sys: 3 ms, total: 11 ms
Wall time: 4.01 s<jupyter_text>Lazy operations in parrallel with distributed<jupyter_code>from dask.distributed import Client
# Fire-up a cluster using the jupyterlab dask extension to your left and insert the appropriate port below:
client = Client("tcp://127.0.0.1:39281", processes=False)
client
%%time
print("z =",z.compute(),"\n")<jupyter_output>z = 3 

CPU times: user 21 ms, sys: 4 ms, total: 25 ms
Wall time: 4.05 s
{'fraction_non_alphanumeric': 0.13592233009708737, 'fraction_numerical': 0.036221060492905154, 'mean_word_length': 4.712153518123667, 'pattern_counts': {'":': 1, '<': 21, '<?xml version=': 0, '>': 21, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 1, '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': '6', 'dir': 'jupyter-structured-clean-dedup', 'id': '475330', 'license': 'MIT', 'n_tokens_mistral': 1069, 'n_tokens_neox': 983, 'n_words': 337, 'path': 'practice/environment/Dask Graph.ipynb', 'repo_name': 'rlambert61/m2poc2019', 'size': '2678'}