code: def test_balance_ray(method, balance, ray_cluster_2cpu):
      config = {
          "input_features": [
              {
                  "name": "Index",
                  "proc_column": "Index",
                  "type": "number",
              },
              {
                  "name": "random_1",
                  "proc_column": "random_1",
                  "type": "number",
              },
              {
                  "name": "random_2",
                  "proc_column": "random_2",
                  "type": "number",
              },
          ],
          "output_features": [
              {
                  "name": "Label",
                  "proc_column": "Label",
                  "type": "binary",
              }
          ],
          "preprocessing": {
              "oversample_minority": None,
              "undersample_majority": None,
          },
      }
      input_df = pd.DataFrame(
          {
              "Index": np.arange(0, 200, 1),
              "random_1": np.random.randint(0, 50, 200),
              "random_2": np.random.choice(
                  ["Type A", "Type B", "Type C", "Type D"], 200
              ),
              "Label": np.concatenate((np.zeros(180), np.ones(20))),
              "split": np.zeros(200),
          }
      )
      config["preprocessing"][method] = balance
      target = config["output_features"][0][NAME]
  
      backend = create_ray_backend()
      input_df = backend.df_engine.from_pandas(input_df)
      test_df = balance_data(
          input_df,
          config["output_features"],
          config["preprocessing"],
          backend,
          42,
      )
  
      majority_class = (
          test_df[target]
          .value_counts()
          .compute()[test_df[target].value_counts().compute().idxmax()]
      )
      minority_class = (
          test_df[target]
          .value_counts()
          .compute()[test_df[target].value_counts().compute().idxmin()]
      )
      new_class_balance = round(minority_class / majority_class, 2)
  
      assert (
          abs(balance - new_class_balance)
          < BALANCE_PERCENTAGE_TOLERANCE
      )
  
imports: import copy
  import os
  import tempfile
  from typing import Literal, Optional, Union
  import numpy
  import pandas
  import pytest
  import torch
  from ludwig.api import LudwigModel
  from ludwig.backend import create_ray_backend, initialize_backend, LOCAL_BACKEND
  from ludwig.constants import AUDIO, BAG, BALANCE_PERCENTAGE_TOLERANCE, BATCH_SIZE, BFILL, BINARY, CATEGORY, COLUMN, DATE, H3, IMAGE, MAX_BATCH_SIZE_DATASET_FRACTION, NAME, NUMBER, PREPROCESSING, SEQUENCE, SET, SPLIT, TEXT, TIMESERIES, TRAINER, VECTOR
  from ludwig.data.preprocessing import balance_data
  from ludwig.data.split import DEFAULT_PROBABILITIES
  from ludwig.utils.data_utils import read_parquet
  from ludwig.utils.misc_utils import merge_dict
  from tests.integration_tests.utils import audio_feature, augment_dataset_with_none, bag_feature, binary_feature, category_feature, create_data_set_to_use, date_feature, generate_data, h3_feature, image_feature, number_feature, RAY_BACKEND_CONFIG, sequence_feature, set_feature, text_feature, timeseries_feature, train_with_backend, vector_feature
  import dask
  import ray
  import ray.exceptions
  from ray.air.config import DatasetConfig
  from ray.data import Dataset, DatasetPipeline
  from ray.train._internal.dataset_spec import DataParallelIngestSpec
  from ludwig.backend.ray import get_trainer_kwargs, RayBackend
  from ludwig.data.dataframe.dask import DaskEngine
label_desc: Use image node_id 24 for calling a global function with example usage: test_balance_ray(method, balance, ray_cluster_2cpu) without return types