code: def _fit(self, df: DataFrame) -> PerPartitionScalarScalerModel:
      partition_key = self.partition_key
      apply_on_cols = self._apply_on_cols()
  
      if partition_key is None:
          rows = df.select(*apply_on_cols).collect()
          assert len(rows) == 1
          per_group_stats = rows[0].asDict()
      else:
          per_group_stats = df.groupBy(partition_key).agg(
              *apply_on_cols
          )
  
      assert per_group_stats is not None
      return self._create_model(per_group_stats)
  
imports: from abc import ABC, abstractmethod
  from typing import Callable, Dict, List, Optional, Union
  from synapse.ml.cyber.utils.spark_utils import ExplainBuilder, HasSetInputCol, HasSetOutputCol
  from pyspark.ml import Estimator, Transformer
  from pyspark.ml.param.shared import HasInputCol, HasOutputCol, Param, Params
  from pyspark.sql import DataFrame, functions, types
label_desc: Use image node_id 5 for calling the PerPartitionScalarScalerEstimator obj's underlying member method code with example usage: obj._fit(df) and returns: self