class_docstr: This is the AccessAnomaly, a pyspark.ml.Estimator which
  creates the AccessAnomalyModel which is a pyspark.ml.Transformer
code: def _fit(self, df: DataFrame) -> AccessAnomalyModel:
      # index the user and resource columns to allow running the spark ALS algorithm
      the_indexer = indexers.MultiIndexer(
          indexers=[
              indexers.IdIndexer(
                  input_col=self.user_col,
                  partition_key=self.tenant_col,
                  output_col=self.indexed_user_col,
                  reset_per_partition=self.separate_tenants,
              ),
              indexers.IdIndexer(
                  input_col=self.res_col,
                  partition_key=self.tenant_col,
                  output_col=self.indexed_res_col,
                  reset_per_partition=self.separate_tenants,
              ),
          ],
      )
  
      the_indexer_model = the_indexer.fit(df)
  
      # indexed_df is the dataframe with the indices for user and resource
      indexed_df = the_indexer_model.transform(df)
      enriched_df = self._enrich_and_normalize(indexed_df).cache()
  
      user_res_feature_vector_mapping_df = (
          self.create_spark_model_vectors_df(
              enriched_df,
          )
      )
      user_res_norm_cf_df_model = ModelNormalizeTransformer(
          enriched_df,
          self.rank_param,
      ).transform(user_res_feature_vector_mapping_df)
  
      # convert user and resource indices back to names
      user_index_model = the_indexer_model.get_model_by_input_col(
          self.user_col
      )
      res_index_model = the_indexer_model.get_model_by_input_col(
          self.res_col
      )
      assert (
          user_index_model is not None and res_index_model is not None
      )
  
      norm_user_mapping_df = (
          user_res_norm_cf_df_model.user_feature_vector_mapping_df
      )
      norm_res_mapping_df = (
          user_res_norm_cf_df_model.res_feature_vector_mapping_df
      )
  
      indexed_user_col = self.indexed_user_col
      indexed_res_col = self.indexed_res_col
  
      # do the actual index to name mapping (using undo_transform)
      final_user_mapping_df = user_index_model.undo_transform(
          norm_user_mapping_df,
      ).drop(indexed_user_col)
      final_res_mapping_df = res_index_model.undo_transform(
          norm_res_mapping_df
      ).drop(
          indexed_res_col,
      )
  
      tenant_col, user_col, res_col = (
          self.tenant_col,
          self.user_col,
          self.res_col,
      )
  
      history_access_df = self.history_access_df
      access_df = (
          history_access_df
          if history_access_df is not None
          else df.select(tenant_col, user_col, res_col).cache()
      )
  
      (
          user2component_mappings_df,
          res2component_mappings_df,
      ) = ConnectedComponents(
          tenant_col,
          user_col,
          res_col,
      ).transform(
          access_df
      )
  
      return AccessAnomalyModel(
          _UserResourceFeatureVectorMapping(
              tenant_col=self.tenant_col,
              user_col=self.user_col,
              user_vec_col=self.user_vec_col,
              res_col=self.res_col,
              res_vec_col=self.res_vec_col,
              history_access_df=history_access_df,
              user2component_mappings_df=user2component_mappings_df,
              res2component_mappings_df=res2component_mappings_df,
              user_feature_vector_mapping_df=final_user_mapping_df.cache(),
              res_feature_vector_mapping_df=final_res_mapping_df.cache(),
          ),
          self.output_col,
      )
  
imports: import os
  from typing import List, Optional, Tuple
  from synapse.ml.cyber.anomaly.complement_access import ComplementAccessTransformer
  from synapse.ml.cyber.feature import indexers, scalers
  from synapse.ml.cyber.utils import spark_utils
  import numpy
  from pyspark import SQLContext
  from pyspark.ml import Estimator, Transformer
  from pyspark.ml.param.shared import Param, Params
  from pyspark.ml.recommendation import ALS
  from pyspark.sql import DataFrame, functions, types
label_desc: Use image node_id 11 for calling the AccessAnomaly obj's underlying member method code with example usage: obj._fit(df) and returns: AccessAnomalyModel