code: def __init__(
      self,
      scheduler_host=None,
      scatter=None,
      client=None,
      loop=None,
      wait_for_workers_timeout=10,
      **submit_kwargs,
  ):
      super().__init__()
  
      if distributed is None:
          msg = (
              "You are trying to use 'dask' as a joblib parallel backend "
              "but dask is not installed. Please install dask "
              "to fix this error."
          )
          raise ValueError(msg)
  
      if client is None:
          if scheduler_host:
              client = Client(
                  scheduler_host, loop=loop, set_as_default=False
              )
          else:
              try:
                  client = get_client()
              except ValueError as e:
                  msg = (
                      "To use Joblib with Dask first create a Dask Client"
                      "\n\n"
                      "    from dask.distributed import Client\n"
                      "    client = Client()\n"
                      "or\n"
                      "    client = Client('scheduler-address:8786')"
                  )
                  raise ValueError(msg) from e
  
      self.client = client
  
      if scatter is not None and not isinstance(scatter, (list, tuple)):
          raise TypeError(
              "scatter must be a list/tuple, got "
              "`%s`" % type(scatter).__name__
          )
  
      if scatter is not None and len(scatter) > 0:
          # Keep a reference to the scattered data to keep the ids the same
          self._scatter = list(scatter)
          scattered = self.client.scatter(scatter, broadcast=True)
          self.data_futures = {
              id(x): f for x, f in zip(scatter, scattered)
          }
      else:
          self._scatter = []
          self.data_futures = {}
      self.wait_for_workers_timeout = wait_for_workers_timeout
      self.submit_kwargs = submit_kwargs
      self.waiting_futures = as_completed(
          [], loop=client.loop, with_results=True, raise_errors=False
      )
      self._results = {}
      self._callbacks = {}
  
imports: from __future__ import print_function, division, absolute_import
  import asyncio
  import concurrent.futures
  import contextlib
  import time
  from uuid import uuid4
  import weakref
  from .parallel import parallel_config
  from .parallel import AutoBatchingMixin, ParallelBackendBase
label_desc: Use image node_id 1 to create a new DaskDistributedBackend object from inherited base classes: AutoBatchingMixin, ParallelBackendBase with example: obj = DaskDistributedBackend(scheduler_host, scatter, client, loop, wait_for_workers_timeout)