Named clusters

A dask task executor runs the tasks on a dask cluster. The module gain.dask.named_cluster builds the dask client of that cluster from a named cluster or from one cluster definition. A cluster definition holds the type, the params and the worker count that build one dask cluster. A named cluster is a cluster definition stored under a name. This page describes the Python side of the named clusters.

Creating a dask client from a named cluster

setup_client() takes the name of a named cluster and a worker count. It returns the dask client and the cluster definition that built it:

from gain.dask.named_cluster import setup_client

client, cluster_definition = setup_client("local", number_of_workers=4)

None as the name selects the named cluster in gain_named_cluster.default. None as the worker count keeps the number_of_workers of the cluster definition.

setup_client_from_config() takes one cluster definition mapping, with type, params and a worker count. It needs no named cluster file. The cluster definition below starts two workers that are threads of the current process, with no dashboard:

# Two workers that are threads of this process, with no dashboard.
LOW_COST_CLUSTER: dict[str, Any] = {
    "type": "local",
    "number_of_workers": 2,
    "params": {"processes": False, "dashboard_address": None},
}

client, _ = setup_client_from_config(LOW_COST_CLUSTER)

setup_client_from_config works on its own copy of the mapping that it gets. The mapping stays unchanged, so one cluster definition builds many clients.

The lifecycle

Give the client to a DaskExecutor. The task executor owns the client from then on. When the task executor closes, it shuts down the client and its cluster. Do not use the client after that.

Where the line falls

The named cluster file, the four types local, sge, slurm and manual, and the options -N and -c are the user side. The user page Named clusters describes them, and the user page Running in parallel describes the task graph options. This page describes only the two functions that build a dask client.

A running example, part 4

Part 1, on the page Graphs, builds a graph that counts the N bases of a reference genome. Part 2, on the page Task executors, runs it from Python, and part 3, on the page Running from a CLI, runs it from a command line. Part 4 runs it on a dask cluster.

The first run builds a dask client from the cluster definition above and gives it to a DaskExecutor. The with block closes the task executor, so the client is closed after the run:

def run_on_a_dask_client(resource: GenomicResource, client: Client) -> int:
    """Count the ``N`` bases of the genome on the cluster of ``client``."""
    graph, _ = build_n_count_graph(resource)
    # The task executor owns the client. Closing it shuts down the cluster.
    with DaskExecutor(client) as executor:
        results = list(task_graph_run_with_results(graph, executor))
    total: int = results[-1]
    return total

The second run uses the script of part 3. -N local selects the named cluster local, and the script builds the client with setup_client:

count_n --grr-directory grr -R genome -N local -d task-status

A dask run prints logs that change from run to run, so the page shows no output. After the run, the record of the summary task sum_n in the task status directory task-status holds the sum.

API

setup_client and setup_client_from_config are defined in the module gain.dask.named_cluster.

`setup_client`([cluster_name, number_of_workers])	Create a dask client for a named cluster.
`setup_client_from_config`(cluster_config, *)	Create a dask client from one cluster definition.

setup_client(cluster_name: str | None = None, number_of_workers: int | None = None) → tuple[Client, dict[str, Any]][source]

Create a dask client for a named cluster.

Parameters:

cluster_name – The name of a named cluster under gain_named_cluster.clusters. None selects the named cluster in gain_named_cluster.default.
number_of_workers – The number of workers. None keeps the number in the cluster definition.

Returns:

The dask client and a copy of the cluster definition that built it.

Return type:

tuple[Client, dict[str, Any]]

Raises:

ValueError – When gain_named_cluster.clusters is not a mapping from names to cluster definitions, when a cluster definition holds a name key, or when no named cluster has the name cluster_name. Also when setup_client_from_config() refuses the cluster definition.

Notes

The function does not change the cluster definition in the dask config. Two calls for the same named cluster build two clients with the same settings.

setup_client_from_config(cluster_config: dict[str, Any], *, number_of_workers: int | None = None) → tuple[Client, dict[str, Any]][source]

Create a dask client from one cluster definition.

Parameters:

cluster_config – A cluster definition: a mapping with the keys type, params and, optionally, number_of_workers or adapt_params. type is local, sge, slurm or manual. params holds the arguments of the dask cluster class. For manual, params holds the address of a running dask scheduler.
number_of_workers – The number of workers. A number here wins over the number_of_workers of the cluster definition. None keeps the number of the cluster definition. When the cluster definition has no number_of_workers, the deprecated number_of_threads key gives the number. When no key gives a number, the cluster adapts by adapt_params, if the cluster definition has that key.

Returns:

The dask client and a copy of cluster_config. The copy is equal to cluster_config. It is not the same object.

Return type:

tuple[Client, dict[str, Any]]

Raises:

ValueError – When type is not a known cluster type, or when a manual cluster definition has no address under params.

Notes

For the local type, each worker runs one thread. When params holds threads_per_worker, the function logs a warning, also for the value 1, and uses 1. For sge and slurm, the jobqueue parameters cores and processes in params set the threads of a worker.

The function works on its own copy of cluster_config and does not change the mapping that it gets. Two calls with the same mapping build two clients with the same settings.