class_docstr: A Datasets-based preprocessor that fits scalers/encoders to the training
  dataset and transforms the training, testing, and inference datasets using
  those fitted scalers/encoders.
class_docstr_tok: ["A","Datasets-based","preprocessor","that","fits","scalers\/encoders","to","the","training","dataset","and","transforms","the","training",",","testing",",","and","inference","datasets","using","those","fitted","scalers\/encoders","."]
code: def _preprocess(
      self, ds: ray.data.Dataset, inferencing: bool
  ) -> Tuple[ray.data.Dataset, ray.data.Dataset]:
      print(
          "\nStep 1: Dropping nulls, creating new_col, updating feature_1\n"
      )
  
      def batch_transformer(df: pd.DataFrame):
          # Disable chained assignment warning.
          pd.options.mode.chained_assignment = None
  
          # Drop nulls.
          df = df.dropna(subset=["nullable_feature"])
  
          # Add new column.
          df["new_col"] = (
              df["feature_1"] - 2 * df["feature_2"] + df["feature_3"]
          ) / 3.0
  
          # Transform column.
          df["feature_1"] = 2.0 * df["feature_1"] + 0.1
  
          return df
  
      ds = ds.map_batches(batch_transformer, batch_format="pandas")
  
      print(
          "\nStep 2: Precalculating fruit-grouped mean for new column and "
          "for one-hot encoding (latter only uses fruit groups)\n"
      )
      agg_ds = ds.groupby("fruit").mean("feature_1")
      fruit_means = {
          r["fruit"]: r["mean(feature_1)"] for r in agg_ds.take_all()
      }
  
      print(
          "\nStep 3: create mean_by_fruit as mean of feature_1 groupby "
          "fruit; one-hot encode fruit column\n"
      )
  
      if inferencing:
          assert self.fruits is not None
      else:
          assert self.fruits is None
          self.fruits = list(fruit_means.keys())
  
      fruit_one_hots = {
          fruit: collections.defaultdict(int, fruit=1)
          for fruit in self.fruits
      }
  
      def batch_transformer(df: pd.DataFrame):
          # Add column containing the feature_1-mean of the fruit groups.
          df["mean_by_fruit"] = df["fruit"].map(fruit_means)
  
          # One-hot encode the fruit column.
          for fruit, one_hot in fruit_one_hots.items():
              df[f"fruit_{fruit}"] = df["fruit"].map(one_hot)
  
          # Drop the fruit column, which is no longer needed.
          df.drop(columns="fruit", inplace=True)
  
          return df
  
      ds = ds.map_batches(batch_transformer, batch_format="pandas")
  
      if inferencing:
          print("\nStep 4: Standardize inference dataset\n")
          assert self.standard_stats is not None
      else:
          assert self.standard_stats is None
  
          print(
              "\nStep 4a: Split training dataset into train-test split\n"
          )
  
          # Split into train/test datasets.
          split_index = int(0.9 * ds.count())
          # Split into 90% training set, 10% test set.
          train_ds, test_ds = ds.split_at_indices([split_index])
  
          print(
              "\nStep 4b: Precalculate training dataset stats for "
              "standard scaling\n"
          )
          # Calculate stats needed for standard scaling feature columns.
          feature_columns = [
              col for col in train_ds.schema().names if col != "label"
          ]
          standard_aggs = [
              agg(on=col)
              for col in feature_columns
              for agg in (Mean, Std)
          ]
          self.standard_stats = train_ds.aggregate(*standard_aggs)
          print("\nStep 4c: Standardize training dataset\n")
  
      # Standard scaling of feature columns.
      standard_stats = self.standard_stats
  
      def batch_standard_scaler(df: pd.DataFrame):
          def column_standard_scaler(s: pd.Series):
              if s.name == "label":
                  # Don't scale the label column.
                  return s
              s_mean = standard_stats[f"mean({s.name})"]
              s_std = standard_stats[f"std({s.name})"]
              return (s - s_mean) / s_std
  
          return df.transform(column_standard_scaler)
  
      if inferencing:
          # Apply standard scaling to inference dataset.
          inference_ds = ds.map_batches(
              batch_standard_scaler, batch_format="pandas"
          )
          return inference_ds, None
      else:
          # Apply standard scaling to both training dataset and test dataset.
          train_ds = train_ds.map_batches(
              batch_standard_scaler, batch_format="pandas"
          )
          test_ds = test_ds.map_batches(
              batch_standard_scaler, batch_format="pandas"
          )
          return train_ds, test_ds
  
imports: import argparse
  import collections
  import os
  import sys
  import time
  from tempfile import TemporaryDirectory
  from typing import Tuple
  import boto3
  import mlflow
  import pandas
  import torch
  import torch.nn
  import torch.optim
  import ray
  from ray import train
  from ray.air.integrations.mlflow import MLflowLoggerCallback
  from ray.data.aggregate import Mean, Std
  from ray.train import Checkpoint, DataConfig, RunConfig, ScalingConfig
  from ray.train.torch.torch_trainer import TorchTrainer
label_desc: Use image node_id 4 for calling the DataPreprocessor obj's underlying member method code with example usage: obj._preprocess(ds, inferencing) and returns: df, df, df, inference_ds,  None, train_ds,  test_ds, unknown, s