class_docstr: Fugue Backend for PyCaret
  
  Example
  -------
  
  >>> from pycaret.datasets import get_data
  >>> from pyspark.sql import SparkSession
  >>> from pycaret.parallel import FugueBackend
  >>> juice = get_data('juice')
  >>> from pycaret.classification import *
  >>> exp_name = setup(data = juice,  target = 'Purchase')
  >>> spark = SparkSessiong.builder.getOrCreate()
  >>> best_model = compare_models(parallel=FugueBackend(spark))
  
  
  engine: Any, default = None
      A ``SparkSession`` instance or "spark" to get the current ``SparkSession``.
      "dask" to get the current Dask client. "native" to test locally using single
      thread. None means using Fugue's Native execution engine to run jobs
      sequentially on local machine, it is for testing/development purpose.
  
  conf: Any, default = None
      Fugue ExecutionEngine configs.
  
  batch_size: int, default = 1
      Batch size to partition the tasks. For example if there are 16 tasks,
      and with ``batch_size=3`` we will have 6 batches to be processed
      distributedly. Smaller batch sizes will have better load balance but
      worse overhead, and vise versa.
  
  display_remote: bool, default = False
      Whether show progress bar and metrics table when using a distributed
      backend. By setting it to True, you must also enable the
      `callback settings <https://fugue-project.github.io/tutorials/tutorials/advanced/rpc.html>`_
      to receive realtime updates.
  
  top_only: bool, default = False
      Whether only return the top ``n_select`` models from each worker. When top only,
      the overall execution time can be faster.
class_docstr_tok: ["Fugue","Backend","for","PyCaret","Example","--","--","--","-",">",">",">","from","pycaret.datasets","import","get_data",">",">",">","from","pyspark.sql","import","SparkSession",">",">",">","from","pycaret.parallel","import","FugueBackend",">",">",">","juice","=","get_data","(","'juice","'",")",">",">",">","from","pycaret.classification","import","*",">",">",">","exp_name","=","setup","(","data","=","juice",",","target","=","'Purchase","'",")",">",">",">","spark","=","SparkSessiong.builder.getOrCreate","(",")",">",">",">","best_model","=","compare_models","(","parallel=FugueBackend","(","spark",")",")","engine",":","Any",",","default","=","None","A","``","SparkSession","``","instance","or","``","spark","''","to","get","the","current","``","SparkSession","``",".","``","dask","''","to","get","the","current","Dask","client",".","``","native","''","to","test","locally","using","single","thread",".","None","means","using","Fugue","'s","Native","execution","engine","to","run","jobs","sequentially","on","local","machine",",","it","is","for","testing\/development","purpose",".","conf",":","Any",",","default","=","None","Fugue","ExecutionEngine","configs",".","batch_size",":","int",",","default","=","1","Batch","size","to","partition","the","tasks",".","For","example","if","there","are","16","tasks",",","and","with","``","batch_size=3","``","we","will","have","6","batches","to","be","processed","distributedly",".","Smaller","batch","sizes","will","have","better","load","balance","but","worse","overhead",",","and","vise","versa",".","display_remote",":","bool",",","default","=","False","Whether","show","progress","bar","and","metrics","table","when","using","a","distributed","backend",".","By","setting","it","to","True",",","you","must","also","enable","the","`","callback","settings","<","https",":","\/\/fugue-project.github.io\/tutorials\/tutorials\/advanced\/rpc.html",">","`","_","to","receive","realtime","updates",".","top_only",":","bool",",","default","=","False","Whether","only","return","the","top","``","n_select","``","models","from","each","worker",".","When","top","only",",","the","overall","execution","time","can","be","faster","."]
code: def compare_models(
      self, instance: Any, params: Dict[str, Any]
  ) -> Union[Any, List[Any]]:
      self._params = dict(params)
      assert "include" in self._params
      assert "sort" in self._params
  
      sort_col, asc = instance._process_sort(self._params["sort"])
  
      shuffled_idx = pd.DataFrame(
          dict(
              idx=random.sample(
                  range(len(self._params["include"])),
                  len(self._params["include"]),
              )
          )
      )
      du = _DisplayUtil(
          self._params.get("display", None),
          progress=-1
          if not self._display_remote
          else shuffled_idx.shape[0],
          verbose=self._params.get("verbose", False),
          sort=sort_col,
          asc=asc,
      )
      outputs = transform(
          shuffled_idx,
          self._remote_compare_models,
          schema="output:binary",
          partition={
              "num": ceil(shuffled_idx.shape[0] / self._batch_size),
              "algo": "even",
          },
          engine=self._engine,
          engine_conf=self._conf,
          callback=None if not self._display_remote else du.update,
          as_fugue=True,
          as_local=True,
      ).as_array()
      res = pd.concat(cloudpickle.loads(x[0]) for x in outputs)
      res = res.sort_values(sort_col, ascending=asc)
      top = res.head(self._params.get("n_select", 1))
      instance._display_container.append(res.iloc[:, :-1])
      top_models = [cloudpickle.loads(x) for x in top._model]
      du.finish(res.iloc[:, :-1])
      return top_models[0] if len(top_models) == 1 else top_models
  
imports: import random
  from math import ceil
  from threading import RLock
  from typing import Any, Callable, Dict, List, Optional, Union
  import cloudpickle
  import pandas
  from fugue import transform
  from pycaret.internal.display import CommonDisplay
  from pycaret.internal.parallel.parallel_backend import ParallelBackend
label_desc: Use image node_id 3 for calling the FugueBackend obj's underlying member method code with example usage: obj.compare_models(instance, params) and returns: unknown