class_docstr: Adaptively allocate workers based on scheduler load.  A superclass.
  
  Contains logic to dynamically resize a Dask cluster based on current use.
  This class needs to be paired with a system that can create and destroy
  Dask workers using a cluster resource manager.  Typically it is built into
  already existing solutions, rather than used directly by users.
  It is most commonly used from the ``.adapt(...)`` method of various Dask
  cluster classes.
  
  Parameters
  ----------
  cluster: object
      Must have scale and scale_down methods/coroutines
  interval : timedelta or str, default "1000 ms"
      Milliseconds between checks
  wait_count: int, default 3
      Number of consecutive times that a worker should be suggested for
      removal before we remove it.
  target_duration: timedelta or str, default "5s"
      Amount of time we want a computation to take.
      This affects how aggressively we scale up.
  worker_key: Callable[WorkerState]
      Function to group workers together when scaling down
      See Scheduler.workers_to_close for more information
  minimum: int
      Minimum number of workers to keep around
  maximum: int
      Maximum number of workers to keep around
  **kwargs:
      Extra parameters to pass to Scheduler.workers_to_close
  
  Examples
  --------
  
  This is commonly used from existing Dask classes, like KubeCluster
  
  >>> from dask_kubernetes import KubeCluster
  >>> cluster = KubeCluster()
  >>> cluster.adapt(minimum=10, maximum=100)
  
  Alternatively you can use it from your own Cluster class by subclassing
  from Dask's Cluster superclass
  
  >>> from distributed.deploy import Cluster
  >>> class MyCluster(Cluster):
  ...     def scale_up(self, n):
  ...         """ Bring worker count up to n """
  ...     def scale_down(self, workers):
  ...        """ Remove worker addresses from cluster """
  
  >>> cluster = MyCluster()
  >>> cluster.adapt(minimum=10, maximum=100)
  
  Notes
  -----
  Subclasses can override :meth:`Adaptive.target` and
  :meth:`Adaptive.workers_to_close` to control when the cluster should be
  resized. The default implementation checks if there are too many tasks
  per worker or too little memory available (see
  :meth:`distributed.Scheduler.adaptive_target`).
  The values for interval, min, max, wait_count and target_duration can be
  specified in the dask config under the distributed.adaptive key.
class_docstr_tok: ["Adaptively","allocate","workers","based","on","scheduler","load",".","A","superclass",".","Contains","logic","to","dynamically","resize","a","Dask","cluster","based","on","current","use",".","This","class","needs","to","be","paired","with","a","system","that","can","create","and","destroy","Dask","workers","using","a","cluster","resource","manager",".","Typically","it","is","built","into","already","existing","solutions",",","rather","than","used","directly","by","users",".","It","is","most","commonly","used","from","the","``",".adapt","(","...",")","``","method","of","various","Dask","cluster","classes",".","Parameters","--","--","--","--","--","cluster",":","object","Must","have","scale","and","scale_down","methods\/coroutines","interval",":","timedelta","or","str",",","default","``","1000","ms","''","Milliseconds","between","checks","wait_count",":","int",",","default","3","Number","of","consecutive","times","that","a","worker","should","be","suggested","for","removal","before","we","remove","it",".","target_duration",":","timedelta","or","str",",","default","``","5s","''","Amount","of","time","we","want","a","computation","to","take",".","This","affects","how","aggressively","we","scale","up",".","worker_key",":","Callable","[","WorkerState","]","Function","to","group","workers","together","when","scaling","down","See","Scheduler.workers_to_close","for","more","information","minimum",":","int","Minimum","number","of","workers","to","keep","around","maximum",":","int","Maximum","number","of","workers","to","keep","around","*","*","kwargs",":","Extra","parameters","to","pass","to","Scheduler.workers_to_close","Examples","--","--","--","--","This","is","commonly","used","from","existing","Dask","classes",",","like","KubeCluster",">",">",">","from","dask_kubernetes","import","KubeCluster",">",">",">","cluster","=","KubeCluster","(",")",">",">",">","cluster.adapt","(","minimum=10",",","maximum=100",")","Alternatively","you","can","use","it","from","your","own","Cluster","class","by","subclassing","from","Dask","'s","Cluster","superclass",">",">",">","from","distributed.deploy","import","Cluster",">",">",">","class","MyCluster","(","Cluster",")",":","...","def","scale_up","(","self",",","n",")",":","...","``","''","''","Bring","worker","count","up","to","n","``","''","''","...","def","scale_down","(","self",",","workers",")",":","...","``","''","''","Remove","worker","addresses","from","cluster","``","''","''",">",">",">","cluster","=","MyCluster","(",")",">",">",">","cluster.adapt","(","minimum=10",",","maximum=100",")","Notes","--","--","-","Subclasses","can","override",":","meth",":","`","Adaptive.target","`","and",":","meth",":","`","Adaptive.workers_to_close","`","to","control","when","the","cluster","should","be","resized",".","The","default","implementation","checks","if","there","are","too","many","tasks","per","worker","or","too","little","memory","available","(","see",":","meth",":","`","distributed.Scheduler.adaptive_target","`",")",".","The","values","for","interval",",","min",",","max",",","wait_count","and","target_duration","can","be","specified","in","the","dask","config","under","the","distributed.adaptive","key","."]
code: def observed(self):
      return self.cluster.observed
  
file_path: distributed/distributed/deploy/adaptive.py
imports: from __future__ import annotations
  import logging
  from inspect import isawaitable
  from tornado.ioloop import IOLoop
  import dask.config
  from dask.utils import parse_timedelta
  from distributed.deploy.adaptive_core import AdaptiveCore
  from distributed.protocol import pickle
  from distributed.utils import log_errors
label_desc: Use image node_id 5 for calling the Adaptive obj's underlying member method code with example usage: obj.observed() and returns: self