cars.orchestrator.cluster.abstract_dask_cluster

Contains abstract function for abstract dask Cluster

Classes

`AbstractDaskCluster`	AbstractDaskCluster
`DaskFutureIterator`	iterator on dask futures, similar to as_completed
`ComputeDSMMemoryLogger`	A subclass of WorkerPlugin dedicated to monitoring workers memory

Functions

`set_config`()	Set particular DASK config such as:
`save_config`(output_dir, file_name)	Save DASK global config
`write_yaml_config`(yaml_config, output_dir, file_name)	Writes a YAML config to disk.
`register_xarray`()	Add hook to dask so it correctly estimates memory used by xarray

Module Contents

class cars.orchestrator.cluster.abstract_dask_cluster.AbstractDaskCluster(conf_cluster, out_dir, log_dir, launch_worker=True, data_to_propagate=None)[source]

Bases: cars.orchestrator.cluster.abstract_cluster.AbstractCluster

AbstractDaskCluster

nb_workers

task_timeout

walltime

use_memory_logger

config_name

launch_worker = True

activate_dashboard

python

abstract check_conf(conf)[source]

Check configuration

Parameters:: conf (dict) – configuration to check
Returns:: overloaded configuration
Return type:: dict

abstract start_dask_cluster()[source]: Start dask cluster

create_task_wrapped(func, nout=1)[source]

Create task

Parameters:

func – function
nout – number of outputs

get_delayed_type()[source]: Get delayed type

start_tasks(task_list)[source]

Start all tasks

Parameters:: task_list – task list

scatter(data, broadcast=True)[source]

Distribute data through workers

Parameters:: data – task data

future_iterator(future_list, timeout=None)[source]

Start all tasks

Parameters:: future_list – future_list list

class cars.orchestrator.cluster.abstract_dask_cluster.DaskFutureIterator(future_list, timeout=None)[source]

iterator on dask futures, similar to as_completed Only returns the actual results, delete the future after usage

dask_a_c

prev = None

__iter__()[source]

__next__()[source]

cars.orchestrator.cluster.abstract_dask_cluster.set_config()[source]: Set particular DASK config such as: - scheduler

cars.orchestrator.cluster.abstract_dask_cluster.save_config(output_dir: str, file_name: str)[source]

Save DASK global config

Parameters:

output_dir – output directory path
file_name – output file name

cars.orchestrator.cluster.abstract_dask_cluster.write_yaml_config(yaml_config: dict, output_dir: str, file_name: str)[source]

Writes a YAML config to disk. TODO: put in global file if needed elsewhere than DASK conf save.

Parameters:

yaml_config – YAML config to write
output_dir – output directory path
file_name – output file name

cars.orchestrator.cluster.abstract_dask_cluster.register_xarray()[source]: Add hook to dask so it correctly estimates memory used by xarray

class cars.orchestrator.cluster.abstract_dask_cluster.ComputeDSMMemoryLogger(outdir)[source]

Bases: distributed.diagnostics.plugin.WorkerPlugin

A subclass of WorkerPlugin dedicated to monitoring workers memory

This plugin enables two things:

Additional dask log traces (for each worker internal state change):
- amount of tasks
- associated memory
A numpy data file with memory metrics and timing

outdir

setup(worker)[source]: Associate plugin with a worker :param worker: The worker to associate the plugin with

transition(key, start, finish, **kwargs)[source]: Callback when worker changes internal state