code: def run(self):
      args = self.args
  
      logging.info(f"Running device: {'GPU' if self.gpu else 'CPU'}")
  
      if self.gpu and args.enable_dask_cuda_cluster:
          logging.info("Setting up Dask CUDA Cluster")
          self.setup_dask_cuda_cluster(
              visible_devices=args.dask_cuda_visible_gpu_devices,
              device_spill_frac=args.dask_cuda_gpu_device_spill_frac,
              dask_work_dir=args.output_path,
          )
  
      ddf = self.read_data(args.data_path)
      ddf = self.cast_dtypes(ddf)
      ddf = self.filter_by_user_item_freq(ddf)
  
      if args.persist_intermediate_files:
          ddf = self.persist_intermediate(ddf, "_cache/01/")
  
      if args.eval_data_path:
          eval_ddf = self.read_data(args.eval_data_path)
          eval_ddf = self.cast_dtypes(eval_ddf)
  
      if args.predict_data_path:
          test_ddf = self.read_data(args.predict_data_path)
          test_ddf = self.cast_dtypes(test_ddf)
  
      if args.dataset_split_strategy:
          if args.eval_data_path:
              raise ValueError(
                  "You cannot provide both --eval_data_path and --dataset_split_strategy"
              )
  
          ddf, eval_ddf = self.split_datasets(ddf)
  
          if args.persist_intermediate_files:
              ddf = self.persist_intermediate(ddf, "_cache/02/train/")
              eval_ddf = self.persist_intermediate(
                  eval_ddf, "_cache/02/eval/"
              )
  
      nvt_features = self.generate_nvt_features()
      nvt_workflow_features = nvt.Workflow(
          nvt_features, client=self.dask_cluster_client
      )
  
      nvt_targets = self.generate_nvt_targets()
      nvt_workflow_targets = nvt.Workflow(
          nvt_targets, client=self.dask_cluster_client
      )
  
      logging.info(
          "Fitting/transforming the preprocessing on train set"
      )
  
      output_dataset_path = args.output_path
  
      nvt_outputs_folder = os.path.join(
          output_dataset_path, NVT_OUTPUT_FOLDER
      )
  
      if os.path.exists(nvt_outputs_folder):
          logging.info(
              "The NVTabular output folder already exists and is "
              "being deleted: {nvt_outputs_folder}"
          )
          # Delete Folder code
          shutil.rmtree(nvt_outputs_folder)
  
      train_dataset = nvt.Dataset(ddf, cpu=not self.gpu)
      # Processing features and targets in separate workflows, because
      # targets might not be available for test/predict_dataset
      train_dataset_features = nvt_workflow_features.fit_transform(
          train_dataset
      )
      train_dataset_targets = nvt_workflow_targets.fit_transform(
          train_dataset
      )
  
      output_train_dataset_path = os.path.join(
          output_dataset_path, "train"
      )
      logging.info(
          f"Saving transformed train set: {output_train_dataset_path}"
      )
      train_dataset_preproc = self.merge_dataset_features_values(
          train_dataset_features, train_dataset_targets, "train", args
      )
      train_dataset_preproc.to_parquet(
          output_train_dataset_path,
          output_files=args.output_num_partitions,
      )
  
      if args.eval_data_path or args.dataset_split_strategy:
          logging.info("Transforming the eval set")
          eval_dataset = nvt.Dataset(eval_ddf, cpu=not self.gpu)
          # Processing features and targets in separate workflows, because
          # targets might not be available for test/predict_dataset
          eval_dataset_features = nvt_workflow_features.transform(
              eval_dataset
          )
          eval_dataset_targets = nvt_workflow_targets.transform(
              eval_dataset
          )
  
          output_eval_dataset_path = os.path.join(
              output_dataset_path, "eval"
          )
          logging.info(
              f"Saving transformed eval set: {output_eval_dataset_path}"
          )
          eval_dataset_preproc = self.merge_dataset_features_values(
              eval_dataset_features, eval_dataset_targets, "eval", args
          )
          eval_dataset_preproc.to_parquet(
              output_eval_dataset_path,
              output_files=args.output_num_partitions,
          )
  
      if args.predict_data_path:
          # Adding a dummy target column(s) to the test set to perform
          # target encoding op while this issue is not fixed
          # https://github.com/NVIDIA-Merlin/NVTabular/issues/1840
          for col in args.target_encoding_targets:
              if col not in test_ddf.columns:
                  test_ddf[col] = 0
  
          logging.info("Transforming the predict/test set")
  
          predict_dataset = nvt.Dataset(test_ddf, cpu=not self.gpu)
          new_predict_dataset = nvt_workflow_features.transform(
              predict_dataset
          )
          new_predict_dataset = nvt.Dataset(
              new_predict_dataset.to_ddf().drop(INDEX_TMP_COL, axis=1),
              schema=new_predict_dataset.schema.excluding_by_name(
                  [INDEX_TMP_COL]
              ),
              cpu=not self.gpu,
          )
  
          output_predict_dataset_path = os.path.join(
              output_dataset_path, "predict"
          )
          logging.info(
              f"Saving predict/test set: {output_predict_dataset_path}"
          )
  
          new_predict_dataset.to_parquet(
              output_predict_dataset_path,
              output_files=args.output_num_partitions,
          )
      nvt_save_path = os.path.join(nvt_outputs_folder, "workflow")
      logging.info(f"Saving nvtabular workflow to: {nvt_save_path}")
      nvt_workflow_features.save(nvt_save_path)
  
imports: import gc
  import logging
  import os
  import shutil
  from functools import reduce
  from typing import Optional
  import nvtabular
  from merlin.core.dispatch import HAS_GPU
  from merlin.schema import Tags
  from nvtabular import ops
  from .args_parsing import parse_arguments
label_desc: Use image node_id 13 for calling the PreprocessingRunner obj's underlying member method code with example usage: obj.run() without return types