Update Hadoop instructions

%% 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("tcp://192.168.101.82:34781'")

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:63200b25-92dd-40d3-be4c-da66bd063d5b tags:

## viewing the Dask dashboard

Once you have started a Dask cluster, the dasboard can be accessed through the Dashboard URL, typically the port 8787 and the IP of the host where it is running, e.g.: `https://192.168.102.43:8787/status`. This URL is not directly accessible from outside PIC's internal network.

However, PIC's jupyter service comes with the `jupyter-server-proxy` package which redirects the traffic from `https://jupyter.pic.es/user/{username}/proxy/{PORT}/{PATH}` to the `PORT` of host where the user's jupyterlab server is running.

However, PIC's jupyter service comes with the `jupyter-server-proxy` package which redirects the traffic from `https://jupyter.pic.es/user/{username}/proxy/{PORT}/{PATH}` to the `PORT` of the host where the user's jupyterlab server is running.

So it is possible to connect to the Dask dashboard from outside PIC through the URL: `https://jupyter.pic.es/user/{username}/proxy/{PORT}/status`, e.g. `https://jupyter.pic.es/user/torradeflot/proxy/8787/status`.

So it is possible to connect to the Dask dashboard from outside PIC through the URL: `https://jupyter.pic.es/user/{username}/proxy/{PORT}/status`, e.g. `https://jupyter.pic.es/user/torradeflot/proxy/8787/status`

%% 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

```

%% Output

    WARNING:bokeh.server.util:Host wildcard '*' will allow connections originating from multiple (or possibly all) hostnames or IPs. Use non-wildcard values to restrict access explicitly

%% Cell type:code id:e3a6efad tags:

``` python

cluster

```

%% Output

%% 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:c35159c9-9eeb-4170-b802-283eea920958 tags:

## Workers with GPUs

To assign a GPU to your workers, include the `+RequestGpus=1` argument in the worker setup. This can be done by adding it to the `job_extra_directives` attribute when you instantiate the `HTCondorCluster` class.

%% Cell type:code id:f21e666b-39f6-4f6c-91b9-0540eb17d5f6 tags:

``` python

cluster = HTCondorCluster(

    cores=1, memory='2GB', disk='10 GB',

    job_extra_directives={

        'getenv': 'True',

        '+RequestGpus': 1

    }

)

```

%% Cell type:markdown id:9a9fe241-16c1-4eb6-84d2-20aee52fef68 tags:

or through the [configuration of the jupyterlab extension](#Configuration)

```

labextension:

  factory:

    module: 'dask_jobqueue'

    class: 'HTCondorCluster'

    args: []

    kwargs: {"job_extra_directives": {"getenv": "True", "+RequestGpus": 1}}

```

%% Cell type:markdown id:0f349885-8d33-44e8-8c5c-fadade3c4071 tags:

# Security

**It is of uttermost importance to enable security when lauching a Dask cluster**, otherwise a malicious user can impersonate you by connecting to your cluster and submitting jobs to it. The attacker would then have access to your personal (e.g. SSH keys and private files in your home) and shared data.

## in jupyter.pic.es

In order to enforce security in the communication with Dask clusters, the notebooks started through jupyter.pic.es are populated with the environment variables:

```

DASK_DISTRIBUTED__COMM__REQUIRE_ENCRYPTION=true

DASK_DISTRIBUTED__COMM__TLS__SCHEDULER__CERT=${HOME}/.config/dask/security/cert.pem

DASK_DISTRIBUTED__COMM__TLS__CA_FILE=${HOME}/.config/dask/security/ca_file.pem

DASK_DISTRIBUTED__COMM__TLS__CLIENT__KEY=${HOME}/.config/dask/security/key.pem

DASK_DISTRIBUTED__COMM__TLS__WORKER__CERT=${HOME}/.config/dask/security/cert.pem

DASK_DISTRIBUTED__COMM__TLS__SCHEDULER__KEY=${HOME}/.config/dask/security/key.pem

DASK_DISTRIBUTED__COMM__TLS__WORKER__KEY=${HOME}/.config/dask/security/key.pem

DASK_DISTRIBUTED__COMM__TLS__CLIENT__CERT=${HOME}/.config/dask/security/cert.pem

```

See the [official documentation](https://distributed.dask.org/en/stable/tls.html#tls-ssl) to understand the function of each file.

When starting a Dask cluster through the jupyterlab extension or using the `pic_jupyterhub.dask_condor.SecureHTCondor` module in the standard environment these files will be created at runtime.

There might be issues when trying to communicate with a running Dask cluster if the security is not properly configured. See the [Troubleshooting section](#DaskSecurityTroubleshooting) for details

## in a custom cluster

When launching a cluster from a cell or a job, there are two way of providing encryption.

### Temporary security

Using temporary information stored in memory, the cluster won't be accessible from outside the process that spawned it. For this solution, the `cryptography` packages is needed. You can install it with `conda install cryptography`.

%% Cell type:code id:c726c306-8276-4939-937a-7ef9b77e109b tags:

``` python

from dask_jobqueue import HTCondorCluster

from dask.distributed import Client

from distributed.security import Security

temp_sec = Security.temporary()

temp_sec_cluster = HTCondorCluster(

    cores=1, memory='2GB', disk='10 GB',

    job_extra_directives={'getenv': 'True'},

    security=temp_sec)

temp_sec_cluster.scale(2)

```

%% Output

    /data/jupyter/software/envs/jupyter_8440a3ce4a306902/lib/python3.12/site-packages/dask_jobqueue/core.py:745: UserWarning: Using a temporary security object without explicitly setting a shared_temp_directory: writing temp files to current working directory (/nfs/pic.es/user/t/torradeflot/services-code-samples/computing/dask) instead. You can set this value by using dask for e.g. `dask.config.set({'jobqueue.pbs.shared_temp_directory': '~'})`or by setting this value in the config file found in `~/.config/dask/jobqueue.yaml`

      warnings.warn(

    /data/jupyter/software/envs/jupyter_8440a3ce4a306902/lib/python3.12/site-packages/distributed/node.py:182: UserWarning: Port 8787 is already in use.

    Perhaps you already have a cluster running?

    Hosting the HTTP server on port 46143 instead

      warnings.warn(

%% Cell type:code id:60d9cd01-9c75-4d0d-a5bf-6c50ca72fb23 tags:

``` python

temp_sec_client = Client(temp_sec_cluster, security=temp_sec)

```

%% Cell type:markdown id:d7857eca-b75c-4edc-843d-12b1c3c42e58 tags:

### Encryption using preexisting files

If you already have a set of CA, key, certificate files, you can use them to encrypt the traffic with and within the Dask Cluster.

%% Cell type:code id:7af5c771-0278-4a81-a87b-76d54499e327 tags:

``` python

from dask_jobqueue import HTCondorCluster

from dask.distributed import Client

from distributed.security import Security

# Point this variables to the files in the file system

ca_file = '/path/to/ca/file'

cert_file = '/path/to/certificate/file'

key_file = '/path/to/key/file'

sec = Security(

    require_encryption=True,

    tls_ca_file=ca_file,

    tls_client_cert=cert_file,

    tls_client_key=key_file,

    tls_worker_cert=cert_file,

    tls_worker_key=key_file,

    tls_scheduler_cert=cert_file,

    tls_scheduler_key=key_file

)

secure_cluster = HTCondorCluster(

    cores=1, memory='2GB', disk='10 GB',

    job_extra_directives={'getenv': 'True'},

    security=sec)

secure_cluster.scale(1)

```

%% Output

    /data/jupyter/software/envs/jupyter_8440a3ce4a306902/lib/python3.12/site-packages/distributed/node.py:182: UserWarning: Port 8787 is already in use.

    Perhaps you already have a cluster running?

    Hosting the HTTP server on port 39937 instead

      warnings.warn(

%% Cell type:markdown id:c74519fc-f224-4a84-a493-535adfc24c29 tags:

Then the cluster can be accessed from an independent process

%% Cell type:code id:9019bd0d-725c-4f6e-98df-c8f0e903a0e7 tags:

``` python

client = Client(secure_cluster.scheduler_address, security=sec)

```

%% Cell type:markdown id:3c2c9fbb-27f8-405f-9b59-e99bc4b70dc2 tags:

### Encryption using files generated with SecureHTCondor

In a similar fashion as in the previous section, if you ever started a cluster using the `SecureHTCondor` module availalbe in jupyter's base environment ([how to](#Creating-a-cluster)), a set of files that can be used for SSL encryption will already be available in the folder `${HOME}/.config/dask/security` and can be used as follows.

%% Cell type:code id:ec8eedc8-7fbc-4b2a-9171-27614199d815 tags:

``` python

import pathlib

from dask_jobqueue import HTCondorCluster

from dask.distributed import Client

from distributed.security import Security

user_home = pathlib.Path.home()

security_folder = user_home / '.config' / 'dask' / 'security'

ca_file = str(security_folder / 'ca_file.pem')

cert_file = str(security_folder / 'cert.pem')

key_file = str(security_folder / 'key.pem')

sec = Security(

    require_encryption=True,

    tls_ca_file=ca_file,

    tls_client_cert=cert_file,

    tls_client_key=key_file,

    tls_worker_cert=cert_file,

    tls_worker_key=key_file,

    tls_scheduler_cert=cert_file,

    tls_scheduler_key=key_file

)

secure_cluster = HTCondorCluster(

    cores=1, memory='2GB', disk='10 GB',

    job_extra_directives={'getenv': 'True'},

    security=sec)

secure_cluster.scale(1)

```

%% Output

    /data/jupyter/software/envs/jupyter_8440a3ce4a306902/lib/python3.12/site-packages/distributed/node.py:182: UserWarning: Port 8787 is already in use.

    Perhaps you already have a cluster running?

    Hosting the HTTP server on port 39937 instead

      warnings.warn(

%% Cell type:markdown id:cbf71bd3-4ed8-486d-8fcc-f41b104daa41 tags:

Using the scheduler address and the same security configuration you would be able to connect a client from a completely independent process.

%% Cell type:code id:3f367ad5-7f4f-4bcc-bf66-c2d987080c7e tags:

``` python

client = Client(secure_cluster.scheduler_address, security=sec)

```

%% Cell type:markdown id:98630f16-a2d3-493b-b2df-4adc7fc7248a tags:

# Configuration

There are different ways to configure Dask and the related libraries, check the [official documentation](https://docs.dask.org/en/stable/configuration.html)

## Configuration files

You can add YAML files to `~/.config/dask/` to customize Dask's configuration. Specifics for each of the libraries in the Dask stack:

* [distributed](https://distributed.dask.org/en/stable/): framework for distributed computing

* [jobqueue](https://jobqueue.dask.org/en/latest/clusters-configuration.html): library to scale up Dask clusters to a batch system (e.g. HTCondor)

* [dask-labextension](https://github.com/dask/dask-labextension): jupyterlab extension to be able to manage/monitor Dask Clusters through a GUI in jupyterlab

## Programatic access to configuration

Configuration can be managed through the `dask.config` module. In particular, you can:

* `dask.config.get` or `dask.config.set` specific parameters

* list current configuration: `dask.config.config`

%% 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:

<a name="DaskSecurityTroubleshooting"></a>

## Security

If you are trying to connect to a running Dask cluster and security is no properly configured you may encounter an error such as:

```

Cluster Start Error

Cluster

 failed to start: TLS certificate does not match. Check your security

settings. More info at https://distributed.dask.org/en/latest/tls.html

```

There are different reasons why you can see this type of errors

### SSL files don't exist

If you launch a Dask cluster from a notebook in `jupyter.pic.es` 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.

### SSL files got corrupted or expired

In some cases, the files used for the encryption can get corrupted. These files can be found in `${HOME}/.config/dask/security`. If you think this is the case, remove all the files in this folder and start a new Dask cluster from the jupyterlab extension or the `pic_jupyterhub` module in the main environment so that they can be regenerated.

### SSL configuration mismatch between client and cluster

Make sure that the encripytion configuration is consistent. Check Dask environment variables, configuration files and direct arguments used on both ends.

%% Cell type:markdown id:d556cd0b-1d7c-4943-93c0-1abb9f55d1fd tags:

## Starting Dask Cluster through the extension fails

### Got multiple values for keyword argument asynchronous

There's currently a bug in the extension and the second time you try to start a cluster from it you will see this error. You can still spawn newe clusters from a notebook, but not from the extension.

The only workaraound as of now is to close the jupyterlab server and request a new one.

%% Cell type:markdown id:1d589c7c-aa6b-4932-8400-934647b5d324 tags:

## Debugging

### Redirect the HTCondor worker jobs logs to files

Adding this section in the Dask configuration:

```

jobqueue:

  htcondor:

    log-directory: /path/to/the/folder/where/you/want/the/logs

```

will create `worker-{Id}.{out|err|log}` files in the chosen folder.

### Change log level

```

logging:

  version: 1

  formatters:

    default:

      format: '%(asctime)s %(levelname)-8s %(name)-15s %(message)s'

      datefmt: '%Y-%m-%d %H:%M:%S'

  handlers:

    console:

      class: logging.StreamHandler

      level: DEBUG

      formatter: default

  loggers:

    distributed.worker:

      level: DEBUG

      handlers:

        - console

```

### Single thread synchronous scheduler

It can be difficult to debug a problem when using the parallell computing paradigm. For debugging purposes the scheduler can be configured to be single-threaded and synchronous ([doc](https://docs.dask.org/en/stable/scheduling.html#single-thread) ). With this configuration the processing will run serially in the same thread as the notebook, so the log/print messages will be displayed in the notebook.

**WARNING** Using this approach will dramaticallly reduce speed

%% Cell type:code id:f4dbab01-4ae4-4ecc-b27b-5f1dfd9df1c6 tags:

``` python

import dask.array as da

import dask

N = 10

def verbose_sq(chunk):

    print(f'chunk: {chunk}')

    return chunk**2

with dask.config.set(scheduler='synchronous'):

    x = da.arange(N, chunks=(2,))

    x2 = x.map_blocks(verbose_sq).compute()

print(f'x^2 = {x2}')

```

%% Output

    chunk: []

    chunk: [1]

    chunk: [8 9]

    chunk: [6 7]

    chunk: [4 5]

    chunk: [2 3]

    chunk: [0 1]

    x^2 = [ 0  1  4  9 16 25 36 49 64 81]

%% Cell type:markdown id:f10f22ba-223f-4d33-af00-5f8315526e74 tags:

## Dask task is stuck

### Retire a worker

First you need to get the worker name, e.g. `SecureHTCondor-1`, then you can retire it

%% Cell type:code id:97fd2c39-7fbe-4d24-9e6b-338f1785c011 tags:

``` python

client.retire_workers(names=['SecureHTCondor-1'], close_workers=True)

```

%% Cell type:markdown id:cfc82432-da32-4adf-af70-c5ef10190565 tags:

This will kill the worker job and the task will be reallocated to another worker

%% 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