Repository: sb-ai-lab/replay
Issue: cannot resolve 'user_idx'

## 🐛 Bug

I have 'user_id' column in my data, so when I try to use some of the methods or functions, I get this error:
`pyspark.sql.utils.AnalysisException: cannot resolve 'user_idx'`

### To Reproduce
Examples where 'user_idx' column is required:
`replay/models/base_rec.py", line 321, in _fit_wrap  
users = log.select("user_idx").distinct()`

`
replay/filters.py", line 28, in min_entries
    entries_by_user = data_frame.groupBy("user_idx").count()  # type: ignore
`



Changed files:
replay/scenarios/fallback.py, replay/metrics/coverage.py, replay/scenarios/basescenario.py, replay/utils.py, replay/filters.py

diff --git a/replay/filters.py b/replay/filters.py
index 6342701..86f387f 100644
--- a/replay/filters.py
+++ b/replay/filters.py
@@ -12,26 +12,36 @@ from replay.utils import convert2spark
 from replay.session_handler import State
 
 
-def min_entries(data_frame: AnyDataFrame, num_entries: int) -> DataFrame:
+def filter_by_min_count(
+    data_frame: AnyDataFrame, num_entries: int, group_by: str = "user_idx"
+) -> DataFrame:
     """
-    Remove users with less than ``num_entries`` ratings.
+    Remove entries with entities (e.g. users, items) which are presented in `data_frame`
+    less than `num_entries` times. The `data_frame` is grouped by `group_by` column,
+    which is entry column name, to calculate counts.
 
     >>> import pandas as pd
     >>> data_frame = pd.DataFrame({"user_idx": [1, 1, 2]})
-    >>> min_entries(data_frame, 2).toPandas()
+    >>> filter_by_min_count(data_frame, 2).toPandas()
        user_idx
     0         1
     1         1
+
+    :param data_frame: spark or pandas dataframe to apply filter
+    :param num_entries: minimal number of times the entry should appear in dataset
+        in order to remain
+    :param group_by: entity column, which is used to calculate entity occurrence couns
+    :return: filteder `data_frame`
     """
     data_frame = convert2spark(data_frame)
     input_count = data_frame.count()
-    entries_by_user = data_frame.groupBy("user_idx").count()  # type: ignore
-    remaining_users = entries_by_user.filter(
-        entries_by_user["count"] >= num_entries
-    )[["user_idx"]]
-    data_frame = data_frame.join(
-        remaining_users, on="user_idx", how="inner"
-    )  # type: ignore
+    count_by_group = data_frame.groupBy(group_by).agg(
+        sf.count(group_by).alias(f"{group_by}_temp_count")
+    )
+    remaining_entities = count_by_group.filter(
+        count_by_group[f"{group_by}_temp_count"] >= num_entries
+    ).select(group_by)
+    data_frame = data_frame.join(remaining_entities, on=group_by, how="inner")
     output_count = data_frame.count()
     diff = (input_count - output_count) / input_count
     if diff > 0.5:
@@ -39,31 +49,37 @@ def min_entries(data_frame: AnyDataFrame, num_entries: int) -> DataFrame:
     else:
         logger_level = State().logger.info
     logger_level(
-        "current threshold removes %s%% of data", diff,
+        "current threshold removes %s%% of data",
+        diff,
     )
     return data_frame
 
 
-def min_rating(
-    data_frame: AnyDataFrame, value: float, column="relevance"
+def filter_out_low_ratings(
+    data_frame: AnyDataFrame, value: float, rating_column="relevance"
 ) -> DataFrame:
     """
     Remove records with records less than ``value`` in ``column``.
 
     >>> import pandas as pd
-    >>> data_frame = pd.DataFrame({"relevance": [1, 5, 3, 4]})
-    >>> min_rating(data_frame, 3.5).toPandas()
-       relevance
-    0          5
-    1          4
+    >>> data_frame = pd.DataFrame({"relevance": [1, 5, 3.5, 4]})
+    >>> filter_out_low_ratings(data_frame, 3.5).show()
+    +---------+
+    |relevance|
+    +---------+
+    |      5.0|
+    |      3.5|
+    |      4.0|
+    +---------+
+    <BLANKLINE>
     """
     data_frame = convert2spark(data_frame)
-    data_frame = data_frame.filter(data_frame[column] > value)  # type: ignore
+    data_frame = data_frame.filter(data_frame[rating_column] >= value)
     return data_frame
 
 
 # pylint: disable=too-many-arguments,
-def filter_user_interactions(
+def take_num_user_interactions(
     log: DataFrame,
     num_interactions: int = 10,
     first: bool = True,
@@ -100,7 +116,7 @@ def filter_user_interactions(
 
     Only first interaction:
 
-    >>> filter_user_interactions(log_sp, 1, True).orderBy('user_idx').show()
+    >>> take_num_user_interactions(log_sp, 1, True).orderBy('user_idx').show()
     +--------+--------+---+-------------------+
     |user_idx|item_idx|rel|          timestamp|
     +--------+--------+---+-------------------+
@@ -112,7 +128,7 @@ def filter_user_interactions(
 
     Only last interaction:
 
-    >>> filter_user_interactions(log_sp, 1, False, item_col=None).orderBy('user_idx').show()
+    >>> take_num_user_interactions(log_sp, 1, False, item_col=None).orderBy('user_idx').show()
     +--------+--------+---+-------------------+
     |user_idx|item_idx|rel|          timestamp|
     +--------+--------+---+-------------------+
@@ -122,7 +138,7 @@ def filter_user_interactions(
     +--------+--------+---+-------------------+
     <BLANKLINE>
 
-    >>> filter_user_interactions(log_sp, 1, False).orderBy('user_idx').show()
+    >>> take_num_user_interactions(log_sp, 1, False).orderBy('user_idx').show()
     +--------+--------+---+-------------------+
     |user_idx|item_idx|rel|          timestamp|
     +--------+--------+---+-------------------+
@@ -151,13 +167,13 @@ def filter_user_interactions(
     window = Window().orderBy(*sorting_order).partitionBy(col(user_col))
 
     return (
-        log.withColumn("rank", sf.row_number().over(window))
-        .filter(col("rank") <= num_interactions)
-        .drop("rank")
+        log.withColumn("temp_rank", sf.row_number().over(window))
+        .filter(col("temp_rank") <= num_interactions)
+        .drop("temp_rank")
     )
 
 
-def filter_by_user_duration(
+def take_num_days_of_user_hist(
     log: DataFrame,
     days: int = 10,
     first: bool = True,
@@ -165,7 +181,7 @@ def filter_by_user_duration(
     user_col: str = "user_idx",
 ) -> DataFrame:
     """
-    Get first/last ``days`` of user interactions.
+    Get first/last ``days`` of users' interactions.
 
     >>> import pandas as pd
     >>> from replay.utils import convert2spark
@@ -193,7 +209,7 @@ def filter_by_user_duration(
 
     Get first day:
 
-    >>> filter_by_user_duration(log_sp, 1, True).orderBy('user_idx', 'item_idx').show()
+    >>> take_num_days_of_user_hist(log_sp, 1, True).orderBy('user_idx', 'item_idx').show()
     +--------+--------+---+-------------------+
     |user_idx|item_idx|rel|          timestamp|
     +--------+--------+---+-------------------+
@@ -207,7 +223,7 @@ def filter_by_user_duration(
 
     Get last day:
 
-    >>> filter_by_user_duration(log_sp, 1, False).orderBy('user_idx', 'item_idx').show()
+    >>> take_num_days_of_user_hist(log_sp, 1, False).orderBy('user_idx', 'item_idx').show()
     +--------+--------+---+-------------------+
     |user_idx|item_idx|rel|          timestamp|
     +--------+--------+---+-------------------+
@@ -245,7 +261,7 @@ def filter_by_user_duration(
     )
 
 
-def filter_between_dates(
+def take_time_period(
     log: DataFrame,
     start_date: Optional[Union[str, datetime]] = None,
     end_date: Optional[Union[str, datetime]] = None,
@@ -278,7 +294,7 @@ def filter_between_dates(
     +--------+--------+---+-------------------+
     <BLANKLINE>
 
-    >>> filter_between_dates(log_sp, start_date="2020-01-01 14:00:00", end_date=datetime(2020, 1, 3, 0, 0, 0)).show()
+    >>> take_time_period(log_sp, start_date="2020-01-01 14:00:00", end_date=datetime(2020, 1, 3, 0, 0, 0)).show()
     +--------+--------+---+-------------------+
     |user_idx|item_idx|rel|          timestamp|
     +--------+--------+---+-------------------+
@@ -305,7 +321,7 @@ def filter_between_dates(
     )
 
 
-def filter_by_duration(
+def take_num_days_of_global_hist(
     log: DataFrame,
     duration_days: int,
     first: bool = True,
@@ -338,7 +354,7 @@ def filter_by_duration(
     +--------+--------+---+-------------------+
     <BLANKLINE>
 
-    >>> filter_by_duration(log_sp, 1).show()
+    >>> take_num_days_of_global_hist(log_sp, 1).show()
     +--------+--------+---+-------------------+
     |user_idx|item_idx|rel|          timestamp|
     +--------+--------+---+-------------------+
@@ -348,7 +364,7 @@ def filter_by_duration(
     +--------+--------+---+-------------------+
     <BLANKLINE>
 
-    >>> filter_by_duration(log_sp, 1, first=False).show()
+    >>> take_num_days_of_global_hist(log_sp, 1, first=False).show()
     +--------+--------+---+-------------------+
     |user_idx|item_idx|rel|          timestamp|
     +--------+--------+---+-------------------+
diff --git a/replay/metrics/coverage.py b/replay/metrics/coverage.py
index 1449865..1942ae8 100644
--- a/replay/metrics/coverage.py
+++ b/replay/metrics/coverage.py
@@ -16,7 +16,7 @@ class Coverage(RecOnlyMetric):
 
     * take ``K`` recommendations with the biggest ``relevance`` for each ``user_id``
     * count the number of distinct ``item_id`` in these recommendations
-    * devide it by the number of items in the whole data set
+    * divide it by the number of items in the whole data set
 
     """
 
diff --git a/replay/scenarios/basescenario.py b/replay/scenarios/basescenario.py
index bc51395..d4ff7f2 100644
--- a/replay/scenarios/basescenario.py
+++ b/replay/scenarios/basescenario.py
@@ -6,7 +6,7 @@ from typing import Optional, Union, Iterable, Dict, List, Any, Tuple
 from pyspark.sql import DataFrame
 
 from replay.constants import AnyDataFrame
-from replay.filters import min_entries
+from replay.filters import filter_by_min_count
 from replay.metrics import Metric, NDCG
 from replay.models.base_rec import BaseRecommender
 from replay.utils import convert2spark
@@ -34,7 +34,7 @@ class BaseScenario(BaseRecommender):
         :param item_features: item features ``[item_id, timestamp]`` + feature columns
         :return:
         """
-        hot_data = min_entries(log, self.threshold)
+        hot_data = filter_by_min_count(log, self.threshold, "user_idx")
         self.hot_users = hot_data.select("user_idx").distinct()
         self._fit_wrap(hot_data, user_features, item_features)
         self.cold_model._fit_wrap(log, user_features, item_features)
@@ -74,7 +74,7 @@ class BaseScenario(BaseRecommender):
         log = convert2spark(log)
         users = users or log or user_features or self.fit_users
         users = self._get_ids(users, "user_idx")
-        hot_data = min_entries(log, self.threshold)
+        hot_data = filter_by_min_count(log, self.threshold, "user_idx")
         hot_users = hot_data.select("user_idx").distinct()
         if not self.can_predict_cold_users:
             hot_users = hot_users.join(self.hot_users)
diff --git a/replay/scenarios/fallback.py b/replay/scenarios/fallback.py
index da92c71..44b4141 100644
--- a/replay/scenarios/fallback.py
+++ b/replay/scenarios/fallback.py
@@ -4,7 +4,7 @@ from typing import Optional, Dict, List, Any, Tuple, Union, Iterable
 from pyspark.sql import DataFrame
 
 from replay.constants import AnyDataFrame
-from replay.filters import min_entries
+from replay.filters import filter_by_min_count
 from replay.metrics import Metric, NDCG
 from replay.models import PopRec
 from replay.models.base_rec import BaseRecommender
@@ -51,7 +51,7 @@ class Fallback(BaseRecommender):
         :param item_features: item features ``[item_id, timestamp]`` + feature columns
         :return:
         """
-        hot_data = min_entries(log, self.threshold)
+        hot_data = filter_by_min_count(log, self.threshold, "user_idx")
         self.hot_users = hot_data.select("user_idx").distinct()
         self._fit_wrap(hot_data, user_features, item_features)
         self.fb_model._fit_wrap(log, user_features, item_features)
@@ -90,7 +90,7 @@ class Fallback(BaseRecommender):
         """
         users = users or log or user_features or self.fit_users
         users = self._get_ids(users, "user_idx")
-        hot_data = min_entries(log, self.threshold)
+        hot_data = filter_by_min_count(log, self.threshold, "user_idx")
         hot_users = hot_data.select("user_idx").distinct()
         hot_users = hot_users.join(self.hot_users, on="user_idx")
         hot_users = hot_users.join(users, on="user_idx", how="inner")
diff --git a/replay/utils.py b/replay/utils.py
index d3f824b..4ea4f12 100644
--- a/replay/utils.py
+++ b/replay/utils.py
@@ -456,7 +456,7 @@ def fallback(
     base: DataFrame, fill: DataFrame, k: int, id_type: str = "idx"
 ) -> DataFrame:
     """
-    Fill missing recommendations for users that have less than ``k`` recomended items.
+    Fill missing recommendations for users that have less than ``k`` recommended items.
     Score values for the fallback model may be decreased to preserve sorting.
 
     :param base: base recommendations that need to be completed
@@ -597,7 +597,9 @@ def add_to_date(
 
 
 def process_timestamp_column(
-    dataframe: DataFrame, column_name: str, date_format: Optional[str] = None,
+    dataframe: DataFrame,
+    column_name: str,
+    date_format: Optional[str] = None,
 ) -> DataFrame:
     """
     Convert ``column_name`` column of numeric/string/timestamp type
@@ -628,7 +630,8 @@ def process_timestamp_column(
 
     # datetime in string format
     dataframe = dataframe.withColumn(
-        column_name, sf.to_timestamp(sf.col(column_name), format=date_format),
+        column_name,
+        sf.to_timestamp(sf.col(column_name), format=date_format),
     )
     return dataframe