# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .py
#       format_name: light
#       format_version: '1.5'
#       jupytext_version: 1.14.4
#   kernelspec:
#     display_name: Python [default]
#     language: python
#     name: python3
# ---

# # 2.1 Dask Delayed
#
# ## Delayed Functions
# The Dask function `delayed` is a *decorator function* (see below) that delays the execution of a function when it is called, letting you determine when to actually execute the function's operation at a later time.

# > **NOTE:** A *decorator function* (or just a *decorator*) is a function that takes another function as its argument and returns yet another function.  Effectively, *decorator functions* act as wrapper functions, passing arguments through the wrapper to the wrapped function.
# >
# >Decorators can be applied in two different ways:
# >
# >```python
# >def unwrapped_function(...):
# >    ...
# >
# >wrapped_function = decorator_function(unwrapped_function)
# >```
# >
# >where the `wrapped_function` is defined separately from the `unwrapped_function`, or
# >
# >```python
# >@decorator_function
# >def function(...):
# >    ...
# >```
# >
# >where the `function` is wrapped at the time it is defined using *decorator syntax*.

from code.mysleep import sleep
import dask


# ## Example: *Slow Python Functions*

# A simple function to increment an integer...slowly!
def slow_inc(x):
    sleep(1)
    return x + 1


# A simple function to decrement an integer...slowly!
def slow_dec(x):
    sleep(1)
    return x - 1


# %time i_2 = slow_inc(2)
i_2

# %time d_i_2 = slow_dec(i_2)
d_i_2

# ## Example: *Dask Delayed Functions*

delayed_inc = dask.delayed(slow_inc)
delayed_dec = dask.delayed(slow_dec)

# %time delayed_i_2 = delayed_inc(2)
delayed_i_2

# %time delayed_d_i_2 = delayed_dec(delayed_i_2)
delayed_d_i_2

# ## Notice anything different?
#
# **1. Run Times:**  The "tasks" ran almost instantaneously!
#
# **2. Return Values:**  The `delayed` functions returned `Delayed` objects.

# ## Delayed Objects
#
# When called, every `delayed` function returns a `Delayed` object.  Each `Delayed` object represents a node in a *task graph*, and each `Delayed` object gives you the ability to examine and visualize the *task graph* that leads up to that node in the graph.
#
# ![](images/inc-add.svg)

# ## Delayed.compute()
#
# To force the `delayed` functions to compute and return the result, we call the `compute` method of the `Delayed` object.

# %time delayed_i_2.compute()

# %time delayed_d_i_2.compute()

# ### Notice!
#
# The computation of `delayed_d_i_2` took 2 seconds, which is the time required to compute `slow_inc(2)` plus the time required to compute `slow_dec(3)`!
#
# But we already computed `delayed_i_2`, so why are we computing it, again?

# ### NOTE:
#
# In addition to using the `compute` method of a Delayed object, you can also compute a `Delayed` object with the `compute` function in Dask.

# %time _i_2, _d_i_2 = dask.compute(delayed_i_2, delayed_d_i_2)
_i_2, _d_i_2

# ### Notice!
#
# Did you notice that this computed both `Delayed` objects at the same time (in parallel)?

# ## Delayed.persist()
#
# To keep the computed result of a `Delayed` object in memory, so that it is available later, we use the `persist` method of the `Delayed` objects.

# %time persist_i_2 = delayed_inc(2).persist()
persist_i_2

# %time persist_i_2.compute()

# %time persist_d_i_2 = delayed_dec(persist_i_2)
persist_d_i_2

# %time persist_d_i_2.compute()

# ### Notice!
#
# Now, the computation of `i2` only took as long as it took to compute `dec(3)` because the result of `i1` was persisted in memory.

# ### NOTE:
#
# Like the `dask.compute` function, you can also persist `Delayed` objects with the `dask.persist` function:

# %time _i_2, _d_i_2 = dask.persist(delayed_i_2, delayed_d_i_2)
_i_2, _d_i_2

# ## Delayed.key
#
# Each `Delayed` object has a unique identifier, called a `key`, which can be returned with the `key` attribute.

delayed_i_2.key

persist_i_2.key

delayed_d_i_2.key

persist_d_i_2.key


# ## Delayed.dask
#
# These `key`s are used to uniquely identify each task in a *Task Graph*, and the *Task Graph* can be viewed as dictionary-like object associated with the `dask` attribute of the `Delayed` object.

# Short function to print out a Task Graph
def print_dask(dobj):
    for key in dobj.dask:
        print('{}:'.format(key))
        if isinstance(dobj.dask[key], tuple):
            print('    function:  {}'.format(dobj.dask[key][0]))
            print('    arguments: {}'.format(dobj.dask[key][1:]))
        else:
            print('    value: {}'.format(dobj.dask[key]))


print_dask(delayed_i_2)

print_dask(delayed_d_i_2)

# ## Delayed.visualize()
#
# It's kinda annoying that we have to write a special function to see what the graph looks like!
#
# Fortunately, there's a better way!  Use the `visualize` method of the `Delayed` object.

delayed_i_2.visualize()

delayed_d_i_2.visualize()

# ### Notice!
#
# If we visualize the persisted versions of these `Delayed` objects, what do you get?

persist_i_2.visualize()

persist_d_i_2.visualize()

# The first objects in the Task Graphs are *data*, now!  Before, they were function calls!

StarcoderdataJupyterScriptsDedupFiltered