class_docstr: Multi-node Multi-GPU K-Nearest Neighbors Classifier Model.
  
  K-Nearest Neighbors Classifier is an instance-based learning technique,
  that keeps training samples around for prediction, rather than trying
  to learn a generalizable set of model parameters.
  
  Parameters
  ----------
  n_neighbors : int (default=5)
      Default number of neighbors to query
  batch_size: int (optional, default 2000000)
      Maximum number of query rows processed at once. This parameter can
      greatly affect the throughput of the algorithm. The optimal setting
      of this value will vary for different layouts and index to query
      ratios, but it will require `batch_size * n_features * 4` bytes of
      additional memory on each worker hosting index partitions.
  handle : cuml.Handle
      Specifies the cuml.handle that holds internal CUDA state for
      computations in this model. Most importantly, this specifies the CUDA
      stream that will be used for the model's computations, so users can
      run different models concurrently in different streams by creating
      handles in several streams.
      If it is None, a new one is created.
  verbose : int or boolean, default=False
      Sets logging level. It must be one of `cuml.common.logger.level_*`.
      See :ref:`verbosity-levels` for more info.
docstr: Predict labels for a query from previously stored index
  and index labels.
  The process is done in a multi-node multi-GPU fashion.
  
  Parameters
  ----------
  X : array-like (device or host) shape = (n_samples, n_features)
      Query data.
      Acceptable formats: dask cuDF, dask CuPy/NumPy/Numba Array
  
  convert_dtype : bool, optional (default = True)
      When set to True, the predict method will automatically
      convert the data to the right formats.
  
  Returns
  -------
  predictions : Dask futures or Dask CuPy Arrays
docstr_tok: ["Predict","labels","for","a","query","from","previously","stored","index","and","index","labels",".","The","process","is","done","in","a","multi-node","multi-GPU","fashion",".","Parameters","--","--","--","--","--","X",":","array-like","(","device","or","host",")","shape","=","(","n_samples",",","n_features",")","Query","data",".","Acceptable","formats",":","dask","cuDF",",","dask","CuPy\/NumPy\/Numba","Array","convert_dtype",":","bool",",","optional","(","default","=","True",")","When","set","to","True",",","the","predict","method","will","automatically","convert","the","data","to","the","right","formats",".","Returns","--","--","--","-","predictions",":","Dask","futures","or","Dask","CuPy","Arrays"]
code: def predict(self, X, convert_dtype=True):
      """
      Predict labels for a query from previously stored index
      and index labels.
      The process is done in a multi-node multi-GPU fashion.
  
      Parameters
      ----------
      X : array-like (device or host) shape = (n_samples, n_features)
          Query data.
          Acceptable formats: dask cuDF, dask CuPy/NumPy/Numba Array
  
      convert_dtype : bool, optional (default = True)
          When set to True, the predict method will automatically
          convert the data to the right formats.
  
      Returns
      -------
      predictions : Dask futures or Dask CuPy Arrays
      """
      query_handler = DistributedDataHandler.create(
          data=X, client=self.client
      )
      self.datatype = query_handler.datatype
  
      comms = KNeighborsClassifier._build_comms(
          self.data_handler, query_handler, self.streams_per_handle
      )
  
      worker_info = comms.worker_info(comms.worker_addresses)
  
      """
          Build inputs and outputs
          """
      self.data_handler.calculate_parts_to_sizes(comms=comms)
      query_handler.calculate_parts_to_sizes(comms=comms)
  
      data_parts_to_ranks, data_nrows = parts_to_ranks(
          self.client, worker_info, self.data_handler.gpu_futures
      )
  
      query_parts_to_ranks, query_nrows = parts_to_ranks(
          self.client, worker_info, query_handler.gpu_futures
      )
  
      """
          Each Dask worker creates a single model
          """
      key = uuid1()
      models = dict(
          [
              (
                  worker,
                  self.client.submit(
                      self._func_create_model,
                      comms.sessionId,
                      **self.kwargs,
                      workers=[worker],
                      key="%s-%s" % (key, idx),
                  ),
              )
              for idx, worker in enumerate(comms.worker_addresses)
          ]
      )
  
      """
          Invoke knn_classify on Dask workers to perform distributed query
          """
      key = uuid1()
      knn_clf_res = dict(
          [
              (
                  worker_info[worker]["rank"],
                  self.client.submit(
                      self._func_predict,
                      models[worker],
                      self.data_handler.worker_to_parts[worker]
                      if worker in self.data_handler.workers
                      else [],
                      data_parts_to_ranks,
                      data_nrows,
                      query_handler.worker_to_parts[worker]
                      if worker in query_handler.workers
                      else [],
                      query_parts_to_ranks,
                      query_nrows,
                      self.uniq_labels,
                      self.n_unique,
                      X.shape[1],
                      worker_info[worker]["rank"],
                      convert_dtype,
                      False,
                      key="%s-%s" % (key, idx),
                      workers=[worker],
                  ),
              )
              for idx, worker in enumerate(comms.worker_addresses)
          ]
      )
  
      wait_and_raise_from_futures(list(knn_clf_res.values()))
  
      """
          Gather resulting partitions and return result
          """
      out_futures = flatten_grouped_results(
          self.client, query_parts_to_ranks, knn_clf_res
      )
      comms.destroy()
  
      return to_output(out_futures, self.datatype).squeeze()
  
imports: from cuml.internals.safe_imports import gpu_only_import
  from cuml.dask.common.input_utils import DistributedDataHandler
  from cuml.dask.common.input_utils import to_output
  from cuml.dask.common import parts_to_ranks
  from cuml.dask.common import flatten_grouped_results
  from cuml.dask.common.utils import raise_mg_import_exception
  from cuml.dask.common.utils import wait_and_raise_from_futures
  from raft_dask.common.comms import get_raft_comm_state
  from cuml.dask.neighbors import NearestNeighbors
  from dask.dataframe import Series
  from dask.distributed import get_worker
  import dask.array
  from uuid import uuid1
  from cuml.internals.safe_imports import cpu_only_import
label_desc: Use image node_id 5 for calling the KNeighborsClassifier obj's underlying member method code with example usage: obj.predict(X, convert_dtype) and returns: to_output