def METHOD_NAME(
    self,
    log: DataFrame,
    user_features: Optional[DataFrame] = None,
    item_features: Optional[DataFrame] = None,
) -> None:
    kmeans = KMeans().setK(self.num_clusters).setFeaturesCol("features")
    user_features_vector = self._transform_features(user_features)
    self.model = kmeans.fit(user_features_vector)
    users_clusters = (
        self.model.transform(user_features_vector)
        .select("user_idx", "prediction")
        .withColumnRenamed("prediction", "cluster")
    )
    log = log.join(users_clusters, on="user_idx", how="left")
    self.item_rel_in_cluster = log.groupBy(["cluster", "item_idx"]).agg(
        sf.count("item_idx").alias("item_count")
    )
    max_count_per_cluster = self.item_rel_in_cluster.groupby(
        "cluster"
    ).agg(sf.max("item_count").alias("max_count_in_cluster"))
    self.item_rel_in_cluster = self.item_rel_in_cluster.join(
        max_count_per_cluster, on="cluster"
    )
    self.item_rel_in_cluster = self.item_rel_in_cluster.withColumn(
        "relevance", sf.col("item_count") / sf.col("max_count_in_cluster")
    ).drop("item_count", "max_count_in_cluster")
    self.item_rel_in_cluster.cache().count()