code: def run_preprocessing(
      tmpdir,
      df_engine,
      input_features,
      output_features,
      dataset_type="parquet",
      num_examples_per_split=20,
      nan_percent=0.0,
      first_row_none=False,
      last_row_none=False,
      nan_cols=None,
  ):
      # Split the dataset manually to avoid randomness in splitting
      split_to_df = {}
      for split in range(3):
          csv_filename = os.path.join(tmpdir, f"{split}_dataset.csv")
          dataset_csv_path = generate_data(
              input_features,
              output_features,
              csv_filename,
              num_examples=num_examples_per_split,
          )
          dataset_df = pd.read_csv(dataset_csv_path)
          dataset_df[SPLIT] = split
          dataset_df.to_csv(dataset_csv_path, index=False)
          split_to_df[split] = dataset_df
      full_df_path = os.path.join(tmpdir, "dataset.csv")
      pd.concat(split_to_df.values()).to_csv(full_df_path, index=False)
      dataset = create_data_set_to_use(
          dataset_type, full_df_path, nan_percent=nan_percent
      )
      dataset = augment_dataset_with_none(
          dataset, first_row_none, last_row_none, nan_cols
      )
  
      # Configure ray backend
      config = {
          "input_features": input_features,
          "output_features": output_features,
          "combiner": {"type": "concat", "output_size": 14},
          TRAINER: {"epochs": 2, "batch_size": 8},
          PREPROCESSING: {
              SPLIT: {
                  "type": "fixed",
              },
          },
      }
      backend_config = {**RAY_BACKEND_CONFIG}
      if df_engine:
          backend_config["processor"]["type"] = df_engine
  
      # Run preprocessing with ray backend
      ray_model = LudwigModel(config, backend=backend_config)
      *ray_datasets, ray_training_set_metadata = ray_model.preprocess(
          skip_save_processed_input=False,  # Save the processed input to test pyarrow write/read
          dataset=dataset,
      )
  
      # Run preprocessing with local backend using the ray_training_set_metadata to ensure parity of
      # token assignments, etc.
      local_model = LudwigModel(config, backend=LOCAL_BACKEND)
      *local_datasets, _ = local_model.preprocess(
          training_set_metadata=ray_training_set_metadata,
          dataset=dataset,
      )
  
      for ray_dataset, local_dataset in zip(
          ray_datasets, local_datasets
      ):
          ray_df = ray_model.backend.df_engine.compute(
              ray_dataset.to_df()
          )
          local_df = local_model.backend.df_engine.compute(
              local_dataset.to_df()
          )
          check_preprocessed_df_equal(local_df, ray_df)
  
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 5 for calling a global function with example usage: run_preprocessing(tmpdir, df_engine, input_features, output_features, dataset_type, num_examples_per_split, nan_percent, first_row_none, last_row_none, nan_cols) without return types