docstr: Starts a Dask CUDA Cluster, so that multiple
  GPUs (memory and compute) can be used for the preprocessing.
  
  Parameters
  ----------
  visible_devices : str
      Comma separated list of visible GPU devices (e.g. "0,1")
  device_spill_frac : float
      Spill GPU-Worker memory to host at this limit.
  dask_work_dir : str
      Local path to use as dask work dir
  Returns
  -------
  Client
      Dask-distributed client
docstr_tok: ["Starts","a","Dask","CUDA","Cluster",",","so","that","multiple","GPUs","(","memory","and","compute",")","can","be","used","for","the","preprocessing",".","Parameters","--","--","--","--","--","visible_devices",":","str","Comma","separated","list","of","visible","GPU","devices","(","e.g",".","``","0,1","''",")","device_spill_frac",":","float","Spill","GPU-Worker","memory","to","host","at","this","limit",".","dask_work_dir",":","str","Local","path","to","use","as","dask","work","dir","Returns","--","--","--","-","Client","Dask-distributed","client"]
code: def setup_dask_cuda_cluster(
      self,
      visible_devices: str = None,
      device_spill_frac: float = 0.7,
      dask_work_dir: Optional[str] = None,
  ):
      """Starts a Dask CUDA Cluster, so that multiple
      GPUs (memory and compute) can be used for the preprocessing.
  
      Parameters
      ----------
      visible_devices : str
          Comma separated list of visible GPU devices (e.g. "0,1")
      device_spill_frac : float
          Spill GPU-Worker memory to host at this limit.
      dask_work_dir : str
          Local path to use as dask work dir
      Returns
      -------
      Client
          Dask-distributed client
      """
      from dask.distributed import Client
      from dask_cuda import LocalCUDACluster
      from merlin.core.utils import device_mem_size
  
      capacity = device_mem_size(
          kind="total"
      )  # Get device memory capacity
      # Reduce if spilling fails to prevent
      # device memory errors.
      cluster = None  # (Optional) Specify existing scheduler port
      if cluster is None:
          cluster = LocalCUDACluster(
              CUDA_VISIBLE_DEVICES="0,1",  # visible_devices,
              local_directory=dask_work_dir,
              device_memory_limit=capacity * device_spill_frac,
          )
  
      # Create the distributed client
      self.dask_cluster_client = Client(cluster)
  
imports: import gc
  import logging
  import os
  import shutil
  from functools import reduce
  from typing import Optional
  import nvtabular
  from merlin.core.dispatch import HAS_GPU
  from merlin.schema import Tags
  from nvtabular import ops
  from .args_parsing import parse_arguments
label_desc: Use image node_id 12 for calling the PreprocessingRunner obj's underlying member method code with example usage: obj.setup_dask_cuda_cluster(visible_devices, device_spill_frac, dask_work_dir) without return types