def recommend_k_items_slow(self, test, top_k=10, remove_seen=True):
        """Recommend top K items for all users which are in the test set.

        Args:
            test: test Spark dataframe
            top_k: top n items to return
            remove_seen: remove items test users have already seen in the past from the recommended set.
        """

        # TODO: remove seen
        if remove_seen:
            raise ValueError("Not implemented")

        self.get_user_affinity(test)\
            .write.mode("overwrite")\
            .saveAsTable(self.f("{prefix}user_affinity"))

        # user_affinity * item_similarity
        # filter top-k
        query = self.f(
            """
        SELECT {col_user}, {col_item}, score
        FROM
        (
          SELECT df.{col_user},
                 S.i2 {col_item},
                 SUM(df.{col_rating} * S.value) AS score,
                 row_number() OVER(PARTITION BY {col_user} ORDER BY SUM(df.{col_rating} * S.value) DESC) rank
          FROM   
            {prefix}user_affinity df, 
            {prefix}item_similarity S
          WHERE df.{col_item} = S.i1
          GROUP BY df.{col_user}, S.i2
        )
        WHERE rank <= {top_k} 
        """,
            top_k=top_k,
        )

        return self.spark.sql(query)
def recommend_k_items_slow(self, test, top_k=10, remove_seen=True):
        """Recommend top K items for all users which are in the test set.

        Args:
            test: test Spark dataframe
            top_k: top n items to return
            remove_seen: remove items test users have already seen in the past from the recommended set.
        """

        # TODO: remove seen
        if remove_seen:
            raise ValueError("Not implemented")

        self.get_user_affinity(test)\
            .write.mode("overwrite")\
            .saveAsTable(self.f("{prefix}user_affinity"))

        # user_affinity * item_similarity
        # filter top-k
        query = self.f(
            """
        SELECT {col_user}, {col_item}, score
        FROM
        (
          SELECT df.{col_user},
                 S.i2 {col_item},
                 SUM(df.{col_rating} * S.value) AS score,
                 row_number() OVER(PARTITION BY {col_user} ORDER BY SUM(df.{col_rating} * S.value) DESC) rank
          FROM   
            {prefix}user_affinity df, 
            {prefix}item_similarity S
          WHERE df.{col_item} = S.i1
          GROUP BY df.{col_user}, S.i2
        )
        WHERE rank <= {top_k} 
        """,
            top_k=top_k,
        )

        return self.spark.sql(query)
def recommend_k_items_slow ( self , test , top_k = 10 , remove_seen = True ) : # TODO: remove seen if remove_seen : raise ValueError ( "Not implemented" ) self . get_user_affinity ( test ) . write . mode ( "overwrite" ) . saveAsTable ( self . f ( "{prefix}user_affinity" ) ) # user_affinity * item_similarity # filter top-k query = self . f ( """
        SELECT {col_user}, {col_item}, score
        FROM
        (
          SELECT df.{col_user},
                 S.i2 {col_item},
                 SUM(df.{col_rating} * S.value) AS score,
                 row_number() OVER(PARTITION BY {col_user} ORDER BY SUM(df.{col_rating} * S.value) DESC) rank
          FROM   
            {prefix}user_affinity df, 
            {prefix}item_similarity S
          WHERE df.{col_item} = S.i1
          GROUP BY df.{col_user}, S.i2
        )
        WHERE rank <= {top_k} 
        """ , top_k = top_k , ) return self . spark . sql ( query )
Recommend top K items for all users which are in the test set.

        Args:
            test: test Spark dataframe
            top_k: top n items to return
            remove_seen: remove items test users have already seen in the past from the recommended set.
Recommend top K items for all users which are in the test set .