<jupyter_start><jupyter_text>Multithreaded Cityblock distance matrix function with SciPy and Dask's delayed executionIn this notebook we the function `scipy.spatial.distance.cdist` to compute the cityblock distance matrix. Althought this function is quite fast, it uses a single thread.In cases like this one, it might be convenient to implement a multithreaded version of the function by parallelicing the execution over chunks of data.We have already written the chunk-based computation on this notebook, but it is missing the parallelization. This chunk-based calculation is kind of pointless wihtout the parallelization: Use `dask.delayed` to compute all chunks in parallel and speed up the calculation.The notebook has no indications of where the modifications need to be done. Just follow the cells and identify what needs to be changed.<jupyter_code>import numpy as np
from scipy.spatial.distance import cdist
from dask import compute, delayed, visualize
nsamples = 12000
nfeat = 50

x = 10. * np.random.random([nsamples, nfeat])<jupyter_output><empty_output><jupyter_text>Let's time the `cdist` function and look the `top` command.<jupyter_code># observe here that the funcion `cdist` used to get the cityblock distance
# is not multithreaded

%timeit cdist(x, x, 'cityblock')<jupyter_output>3.35 s ± 2.72 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)<jupyter_text>With the `top` command we see that `cdist` runs in a single thread. In such cases it could be quite simple write a distributed version of the function. We can do this very easily with `dask.delayed`!  Dask's async delayed executionA simple distributed version of `cdist` can be done as the following:  * Split the array of vectors into chunks. We can use `np.split(x, num_chunks)`  * Compute partial cityblock distance matrices of the complete array with respect to each of the chunks  * Concatenate the resulting list into a single cityblock distance matrix.Note that concatenation is not a fast operation, so probably we will have to continue improving our function.<jupyter_code>np.split?
# define the list of operations to be performed asynchronously
chunks = 12  # we choose on chunk for physical cpu (gpu partition)
partial_distances = [delayed(cdist)(x, xi, 'cityblock') for xi in np.split(x, chunks)]
# visualize the computational graph until this point
visualize(partial_distances)
cbdm = delayed(np.concatenate)(partial_distances, axis=1)
# visualize the computational graph
cbdm.visualize()<jupyter_output><empty_output><jupyter_text>At this point, you should have the computational graph already defined. Let's run and time the compute step. We may go a shell and run the command `top`. Now you should see that the computation is executed in parallel resulting in a shorter execution time.<jupyter_code>#time and run the computational graph
%timeit -n 10 cbdm.compute(scheduler="threads")
# check that the resulting matrices are the same
np.abs(cbdm.compute(scheduler="threads") - cdist(x, x, 'cityblock')).max()<jupyter_output><empty_output><jupyter_text>A problem with this solution, as mentioned above, is that `np.concatenate` is not  a fast operation.Let's check how much time it takes without the concatenation part:<jupyter_code># time and run the computational graph without the concatenate part
%timeit compute(partial_distances, scheduler="threads")<jupyter_output>347 ms ± 12.8 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)<jupyter_text>Let's implement the whole thing as a single function:<jupyter_code># Implementing the whole thing as a single function
def cityblock_dask_concat(x, y, chunks):
    """Implementation using array concatenation"""
# check that the resulting matrices are the same
# print(np.abs(cityblock_dask_concat(x, x, chunks) - cdist(x, x, 'cityblock')).max())<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.07060669456066945, 'fraction_numerical': 0.007845188284518828, 'mean_word_length': 5.6521739130434785, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '6', 'dir': 'jupyter-structured-clean-dedup', 'id': '639961', 'license': 'BSD-3-Clause', 'n_tokens_mistral': 1107, 'n_tokens_neox': 1053, 'n_words': 491, 'path': 'dask/02-exercise-cityblock-distance-matrix-scipy.dask.ipynb', 'repo_name': 'edoardob90/PythonHPC', 'size': '3824'}