code: def transform(self, df: DataFrame) -> Tuple[DataFrame, DataFrame]:
      edges = (
          df.select(self.tenant_col, self.user_col, self.res_col)
          .distinct()
          .orderBy(self.tenant_col, self.user_col, self.res_col)
          .cache()
      )
  
      users = (
          df.select(self.tenant_col, self.user_col)
          .distinct()
          .orderBy(self.tenant_col, self.user_col)
          .cache()
      )
      user2index = spark_utils.DataFrameUtils.zip_with_index(
          users,
          col_name="user_component",
      )
      user2components = user2index
      res2components = None
  
      chg = True
  
      while chg:
          res2components = (
              edges.join(
                  user2components, [self.tenant_col, self.user_col]
              )
              .groupBy(self.tenant_col, self.res_col)
              .agg(f.min("user_component").alias("res_component"))
          )
  
          next_user2components = (
              edges.join(
                  res2components, [self.tenant_col, self.res_col]
              )
              .groupBy(self.tenant_col, self.user_col)
              .agg(f.min("res_component").alias("user_component"))
              .cache()
          )
  
          chg = (
              user2components.join(
                  next_user2components,
                  on=[self.tenant_col, self.user_col, "user_component"],
              ).count()
              != user2components.count()
          )
  
          user2components = next_user2components
  
      assert res2components is not None
  
      return (
          user2components.select(
              self.tenant_col,
              self.user_col,
              f.col("user_component").alias(self.component_col_name),
          ).orderBy(self.tenant_col, self.user_col),
          res2components.select(
              self.tenant_col,
              self.res_col,
              f.col("res_component").alias(self.component_col_name),
          ).orderBy(self.tenant_col, self.res_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 2 for calling the ConnectedComponents obj's underlying member method code with example usage: obj.transform(df) and returns: user2components,  res2components