code: def run_test_imbalance_ray(
      tmpdir,
      input_df,
      config,
      balance,
      num_cpus=2,
      num_gpus=None,
  ):
      with ray_start(num_cpus=num_cpus, num_gpus=num_gpus):
          csv_filename = os.path.join(tmpdir, "dataset.csv")
          input_df.to_csv(csv_filename)
          dataset_parquet = create_data_set_to_use(
              "parquet", csv_filename
          )
  
          model = LudwigModel(
              config, backend=RAY_BACKEND_CONFIG, callbacks=None
          )
          output_dir = None
  
          try:
              _, output_dataset, output_dir = model.train(
                  dataset=dataset_parquet,
                  training_set=None,
                  validation_set=None,
                  test_set=None,
                  skip_save_processed_input=True,
                  skip_save_progress=True,
                  skip_save_unprocessed_output=True,
                  skip_save_log=True,
              )
          finally:
              # Remove results/intermediate data saved to disk
              shutil.rmtree(output_dir, ignore_errors=True)
  
          input_train_set = input_df.sample(frac=0.7, replace=False)
          processed_len = output_dataset[0].ds.count()
          processed_target_pos = output_dataset[0].ds.sum(
              on="Label_mZFLky"
          )
          processed_target_neg = output_dataset[
              0
          ].ds.count() - output_dataset[0].ds.sum(on="Label_mZFLky")
          assert len(input_train_set) == 140
          assert (
              0.05
              <= len(input_train_set[input_train_set["Label"] == 1])
              / len(input_train_set)
              <= 0.15
          )
          assert (
              round(processed_target_pos / processed_target_neg, 1)
              == 0.5
          )
          assert (
              model.backend.df_engine.parallelism
              == RAY_BACKEND_CONFIG["processor"]["parallelism"]
          )
          assert isinstance(model.backend, RayBackend)
  
          if balance == "oversample_minority":
              assert len(input_train_set) < processed_len
  
          if balance == "undersample_majority":
              assert len(input_train_set) > processed_len
  
file_path: ludwig/tests/integration_tests/test_class_imbalance_feature.py
imports: import contextlib
  import os
  import shutil
  import numpy
  import pandas
  import pytest
  from ludwig.api import LudwigModel
  from ludwig.backend import LocalBackend
  from tests.integration_tests.utils import create_data_set_to_use, RAY_BACKEND_CONFIG, spawn
label_desc: Use image node_id 2 for calling a global function with example usage: run_test_imbalance_ray(tmpdir, input_df, config, balance, num_cpus, num_gpus) without return types