Add Dask notebook

%% Cell type:markdown id:bd440c63 tags:

# Dask + HTCondor

%% Cell type:markdown id:c471c42a tags:

# Creating a cluster

%% Cell type:markdown id:86cf720e tags:

## using dask-labextension

%% Cell type:markdown id:99343293 tags:

To create a Dask cluster from the extension. Go to the dask-labextension tab and click "+ NEW"

%% Cell type:markdown id:f32e926e tags:

![kk](static/dask_labextension_1.png)

%% Cell type:markdown id:5cdebde9 tags:

You can see the cluster information and have some shortcuts

* "<>" insert a cell with the necessary lines to create a Dask client. Some environment variables have been set in order to be able to use a "regular" Dask Client

* "SCALE" increase/decrease the number of workers

* "SHUTDOWN" shutdown the cluster

%% Cell type:markdown id:7c2597ed tags:

![kk](static/dask_add_cluster_cell.png)

%% Cell type:code id:61fb5ed7 tags:

``` python

from dask.distributed import Client

client = Client("tls://192.168.100.56:45722")

client

```

%% Cell type:code id:5aa4058d tags:

``` python

client.cluster

```

%% Cell type:markdown id:e2e751cf tags:

The encryption of the traffic between the different Dask components is enforced through environment variables

%% Cell type:code id:909b23d9 tags:

``` python

[f'{k}={v}' for k,v in os.environ.items() if 'DASK' in k]

```

%% Cell type:markdown id:b762046d tags:

![kk](static/dask_labextension_2.png)

%% Cell type:markdown id:86bfa20e tags:

## From python

### Using the default environment

This could be a notebook or a script submitted to HTCondor

%% Cell type:code id:24e002d8 tags:

``` python

from pic_jupyterhub.dask_condor import SecureHTCondor

from dask.distributed import Client

```

%% Cell type:code id:cbcfe3a9 tags:

``` python

cluster = SecureHTCondor() #scheduler_options={'dashboard_address':":8788"})

```

%% Cell type:markdown id:43c320bb tags:

The cluster is initially created without workers. We need to scale it.

%% Cell type:code id:e5d95606 tags:

``` python

cluster.scale(2)

```

%% Cell type:code id:ca7188ff tags:

``` python

client = Client(cluster)

```

%% Cell type:code id:b5b4d24f tags:

``` python

cluster.close()

```

%% Cell type:markdown id:90b5c961 tags:

### Using a custom environment

The environment needs to have:

* `dask-jobqueue` to be able to start Dask clusters with HTCondor

* `ipykernel` to use it as a notebook kernel

* `numpy` and `pandas` to be able to create Dask arrays and dataframes

* `bokeh` for the Dask dashboard

`mamba create (-n {env name} | -p {env Path}) dask-jobqueue ipykernel numpy pandas bokeh`

%% Cell type:code id:1ba4781a tags:

``` python

from dask_jobqueue import HTCondorCluster

from dask.distributed import Client

```

%% Cell type:code id:7dd3c523 tags:

``` python

cluster = HTCondorCluster(cores=1, memory='2GB', disk='10 GB',

                         job_extra_directives={'getenv': 'True'}) # needed to propagate the security

```

%% Cell type:code id:e3a6efad tags:

``` python

cluster

```

%% Cell type:code id:e0c0f659 tags:

``` python

cluster.scale(2)

```

%% Cell type:code id:66cf3e55 tags:

``` python

c = Client(cluster)

```

%% Cell type:markdown id:a101a9d8 tags:

Security is inherited from environment variables

%% Cell type:code id:8b92c694 tags:

``` python

cluster.security

```

%% Cell type:markdown id:b0ffb263 tags:

## Connect to existing cluster

%% Cell type:markdown id:3570b4ab tags:

Meaning a cluster that was launched from outside you jupyterlab instance, e.g. an independent HTCondor job or somebody else's cluster.

Check IP address of running job

```

nslookup $(condor_q $JOB_ID -af RemoteHost | cut -d "@" -f 2)

```

%% Cell type:code id:7c0b19a6 tags:

``` python

from dask.distributed import Client

client = Client("tls://192.168.100.5:42166")

client

```

%% Cell type:markdown id:3809e604 tags:

# Troubleshooting

## Compatibility issues

If you try to connect a notebook to a Dask cluster, and the notebook's environment is different from the one used to launch de cluster, you may encounter compatibility issues.

You will tipycally receive a "Mismatched versions found" warning like this:

```

/data/astro/scratch2/torradeflot/envs/dask/lib/python3.11/site-packages/distributed/client.py:1388: VersionMismatchWarning: Mismatched versions found

+---------+----------------+----------------+----------------+

| Package | Client         | Scheduler      | Workers        |

+---------+----------------+----------------+----------------+

| lz4     | 4.3.3          | 4.3.2          | 4.3.2          |

| msgpack | 1.0.7          | 1.0.5          | 1.0.5          |

| numpy   | 1.26.3         | 1.24.3         | 1.24.3         |

| pandas  | 2.2.0          | 2.0.2          | 2.0.2          |

| python  | 3.11.7.final.0 | 3.11.5.final.0 | 3.11.5.final.0 |

| toolz   | 0.12.1         | 0.12.0         | 0.12.0         |

+---------+----------------+----------------+----------------+

  warnings.warn(version_module.VersionMismatchWarning(msg[0]["warning"]))

```

Some mismatches might be blocking, it is recommended to match the major and minor versions. A mismatch in the patch version shouldn't be a problem.

%% Cell type:markdown id:8b2a984a tags:

## Problems with securitization

If you launch a Dask cluster from a notebook but you have never launched a cluster from the dask-labextension or using the `pic_jupyterhub` module, you may encounter a problem because encryption is enforced but the certificates do not exist.

If this is the case, you can launch a cluster using one of these options as shown above. This will generate the certificate files and the subsequent creation of a Dask cluster from a notebook should succeed.

%% Cell type:markdown id:a28e457b tags:

# Examples

## Dask example 1

picked from https://docs.dask.org/en/stable/10-minutes-to-dask.html

%% Cell type:code id:5353d8fe tags:

``` python

import numpy as np

import pandas as pd

import dask.dataframe as dd

import dask.array as da

import dask.bag as db

```

%% Cell type:markdown id:03a01ccf tags:

### DataFrame

%% Cell type:code id:c7a4f7bd tags:

``` python

index = pd.date_range("2021-09-01", periods=2400, freq="1H")

df = pd.DataFrame({"a": np.arange(2400), "b": list("abcaddbe" * 300)}, index=index)

ddf = dd.from_pandas(df, npartitions=10)

ddf

```

%% Cell type:code id:4102f382 tags:

``` python

ddf.divisions

```

%% Cell type:code id:0f0b932d tags:

``` python

ddf.partitions[1]

```

%% Cell type:code id:0c79af96 tags:

``` python

ddf["2000-10-01": "2021-10-09 5:00"].compute()

```

%% Cell type:markdown id:7277eca8 tags:

### Array

%% Cell type:code id:87156672 tags:

``` python

import numpy as np

import dask.array as da

data = np.arange(100_000).reshape(200, 500)

a = da.from_array(data, chunks=(100, 100))

a

```

%% Cell type:code id:eded606c tags:

``` python

a.chunks

```

%% Cell type:code id:9b36946c tags:

``` python

a.blocks[1, 3]

```

%% Cell type:code id:253153f3 tags:

``` python

a[:50, 200]

```

%% Cell type:code id:82717424 tags:

``` python

a[:50, 200].compute()

```

%% Cell type:code id:c0222fbb tags:

``` python

a.mean()

a.mean().compute()

np.sin(a)

np.sin(a).compute()

a.T

a.T.compute()

```

%% Cell type:code id:73831cd0 tags:

``` python

b = a.max(axis=1)[::-1] + 10

```

%% Cell type:code id:6af83c0b tags:

``` python

b[:10].compute()

```

%% Cell type:code id:81b4ad35 tags:

``` python

b.dask

```

%% Cell type:markdown id:cc48cd7c tags:

## Dask example 2

%% Cell type:code id:0f9c1ff3 tags:

``` python

import dask.array as da

x = da.random.random((30_000, 30_000), chunks=(1000, 1000))

x

```

%% Cell type:code id:35ec13c3 tags:

``` python

y = x + x.T

```

%% Cell type:code id:4ae20942 tags:

``` python

y.sum().compute()

```

%% Cell type:code id:11550bf3 tags:

``` python

y[:, :10].compute()

```

%% Cell type:code id:90dbcbc0 tags:

``` python

```

# ---

# jupyter:

#   jupytext:

#     formats: ipynb,py:light

#     text_representation:

#       extension: .py

#       format_name: light

#       format_version: '1.5'

#       jupytext_version: 1.16.2

#   kernelspec:

#     display_name: Python 3 (ipykernel)

#     language: python

#     name: python3

# ---

# # Dask + HTCondor

# # Creating a cluster

# ## using dask-labextension

# To create a Dask cluster from the extension. Go to the dask-labextension tab and click "+ NEW"

# ![kk](static/dask_labextension_1.png)

# You can see the cluster information and have some shortcuts

# * "<>" insert a cell with the necessary lines to create a Dask client. Some environment variables have been set in order to be able to use a "regular" Dask Client

# * "SCALE" increase/decrease the number of workers

# * "SHUTDOWN" shutdown the cluster

# ![kk](static/dask_add_cluster_cell.png)

# +

from dask.distributed import Client

client = Client("tls://192.168.100.56:45722")

client

# -

client.cluster

# The encryption of the traffic between the different Dask components is enforced through environment variables

[f'{k}={v}' for k,v in os.environ.items() if 'DASK' in k]

# ![kk](static/dask_labextension_2.png)

# ## From python

#

# ### Using the default environment

#

# This could be a notebook or a script submitted to HTCondor

from pic_jupyterhub.dask_condor import SecureHTCondor

from dask.distributed import Client

cluster = SecureHTCondor() #scheduler_options={'dashboard_address':":8788"})

# The cluster is initially created without workers. We need to scale it.

cluster.scale(2)

client = Client(cluster)

cluster.close()

# ### Using a custom environment

#

# The environment needs to have:

#

# * `dask-jobqueue` to be able to start Dask clusters with HTCondor

# * `ipykernel` to use it as a notebook kernel

# * `numpy` and `pandas` to be able to create Dask arrays and dataframes

# * `bokeh` for the Dask dashboard

#

# `mamba create (-n {env name} | -p {env Path}) dask-jobqueue ipykernel numpy pandas bokeh`

#

from dask_jobqueue import HTCondorCluster

from dask.distributed import Client

cluster = HTCondorCluster(cores=1, memory='2GB', disk='10 GB',

                         job_extra_directives={'getenv': 'True'}) # needed to propagate the security

cluster

cluster.scale(2)

c = Client(cluster)

# Security is inherited from environment variables

cluster.security

# ## Connect to existing cluster

# Meaning a cluster that was launched from outside you jupyterlab instance, e.g. an independent HTCondor job or somebody else's cluster.

#

# Check IP address of running job

# ```

# nslookup $(condor_q $JOB_ID -af RemoteHost | cut -d "@" -f 2)

# ```

# +

from dask.distributed import Client

client = Client("tls://192.168.100.5:42166")

client

# -

# # Troubleshooting

#

# ## Compatibility issues

#

# If you try to connect a notebook to a Dask cluster, and the notebook's environment is different from the one used to launch de cluster, you may encounter compatibility issues.

#

# You will tipycally receive a "Mismatched versions found" warning like this:

#

# ```

# /data/astro/scratch2/torradeflot/envs/dask/lib/python3.11/site-packages/distributed/client.py:1388: VersionMismatchWarning: Mismatched versions found

#

# +---------+----------------+----------------+----------------+

# | Package | Client         | Scheduler      | Workers        |

# +---------+----------------+----------------+----------------+

# | lz4     | 4.3.3          | 4.3.2          | 4.3.2          |

# | msgpack | 1.0.7          | 1.0.5          | 1.0.5          |

# | numpy   | 1.26.3         | 1.24.3         | 1.24.3         |

# | pandas  | 2.2.0          | 2.0.2          | 2.0.2          |

# | python  | 3.11.7.final.0 | 3.11.5.final.0 | 3.11.5.final.0 |

# | toolz   | 0.12.1         | 0.12.0         | 0.12.0         |

# +---------+----------------+----------------+----------------+

#   warnings.warn(version_module.VersionMismatchWarning(msg[0]["warning"]))

# ```

#

# Some mismatches might be blocking, it is recommended to match the major and minor versions. A mismatch in the patch version shouldn't be a problem.

# ## Problems with securitization

#

# If you launch a Dask cluster from a notebook but you have never launched a cluster from the dask-labextension or using the `pic_jupyterhub` module, you may encounter a problem because encryption is enforced but the certificates do not exist.

#

# If this is the case, you can launch a cluster using one of these options as shown above. This will generate the certificate files and the subsequent creation of a Dask cluster from a notebook should succeed.

# # Examples

# ## Dask example 1

#

# picked from https://docs.dask.org/en/stable/10-minutes-to-dask.html

# +

import numpy as np

import pandas as pd

import dask.dataframe as dd

import dask.array as da

import dask.bag as db

# -

# ### DataFrame

index = pd.date_range("2021-09-01", periods=2400, freq="1H")

df = pd.DataFrame({"a": np.arange(2400), "b": list("abcaddbe" * 300)}, index=index)

ddf = dd.from_pandas(df, npartitions=10)

ddf

ddf.divisions

ddf.partitions[1]

ddf["2000-10-01": "2021-10-09 5:00"].compute()

# ### Array

# +

import numpy as np

import dask.array as da

data = np.arange(100_000).reshape(200, 500)

a = da.from_array(data, chunks=(100, 100))

a

# -

a.chunks

a.blocks[1, 3]

a[:50, 200]

a[:50, 200].compute()

a.mean()

a.mean().compute()

np.sin(a)

np.sin(a).compute()

a.T

a.T.compute()

b = a.max(axis=1)[::-1] + 10

b[:10].compute()

b.dask

# ## Dask example 2

import dask.array as da

x = da.random.random((30_000, 30_000), chunks=(1000, 1000))

x

y = x + x.T

y.sum().compute()

y[:, :10].compute()