class_docstr: SAR implementation for PySpark.
code: def _recommend_k_items_fast(
      self,
      test,
      top_k=10,
      remove_seen=True,
      n_user_prediction_partitions=200,
  ):
      assert self.cache_path is not None
  
      # create item id to continuous index mapping
      log.info("sarplus.recommend_k_items 1/3: create item index")
      query = self._format(
          """
              SELECT i1, ROW_NUMBER() OVER(ORDER BY i1)-1 AS idx
              FROM (
                    SELECT DISTINCT i1
                    FROM `{prefix}item_similarity`
                   )
              CLUSTER BY i1
          """
      )
      self.spark.sql(query).write.mode("overwrite").saveAsTable(
          self._format("{prefix}item_mapping")
      )
  
      # map similarity matrix into index space
      query = self._format(
          """
              SELECT a.idx AS i1, b.idx AS i2, is.value
              FROM `{prefix}item_similarity` AS is,
                   `{prefix}item_mapping` AS a,
                   `{prefix}item_mapping` AS b
              WHERE is.i1 = a.i1 AND i2 = b.i1
          """
      )
      self.spark.sql(query).write.mode("overwrite").saveAsTable(
          self._format("{prefix}item_similarity_mapped")
      )
  
      cache_path_output = self.cache_path
      if self.cache_path.startswith("dbfs:"):
          # Databricks DBFS
          cache_path_input = "/dbfs" + self.cache_path[5:]
      elif self.cache_path.startswith("synfs:"):
          # Azure Synapse
          # See https://docs.microsoft.com/en-us/azure/synapse-analytics/spark/synapse-file-mount-api
          cache_path_input = "/synfs" + self.cache_path[6:]
      else:
          cache_path_input = self.cache_path
  
      # export similarity matrix for C++ backed UDF
      log.info(
          "sarplus.recommend_k_items 2/3: prepare similarity matrix"
      )
  
      query = self._format(
          """
              SELECT i1, i2, CAST(value AS DOUBLE) AS value
              FROM `{prefix}item_similarity_mapped`
              ORDER BY i1, i2
          """
      )
      self.spark.sql(query).coalesce(1).write.format(
          "com.microsoft.sarplus"
      ).mode("overwrite").save(cache_path_output)
  
      self.get_user_affinity(test).createOrReplaceTempView(
          self._format("{prefix}user_affinity")
      )
  
      # map item ids to index space
      query = self._format(
          """
              SELECT `{col_user}`, idx, rating
              FROM (
                    SELECT `{col_user}`, b.idx, `{col_rating}` AS rating
                    FROM `{prefix}user_affinity`
                    JOIN `{prefix}item_mapping` AS b
                    ON `{col_item}` = b.i1 
                   )
              CLUSTER BY `{col_user}`
          """
      )
      pred_input = self.spark.sql(query)
  
      schema = StructType(
          [
              StructField(
                  "userID",
                  pred_input.schema[self.header["col_user"]].dataType,
                  True,
              ),
              StructField("itemID", IntegerType(), True),
              StructField("score", FloatType(), True),
          ]
      )
  
      # make sure only the header is pickled
      local_header = self.header
  
      # bridge to python/C++
      @pandas_udf(schema, PandasUDFType.GROUPED_MAP)
      def sar_predict_udf(df):
          # Magic happening here.  The cache_path points to file write
          # to by com.microsoft.sarplus This has exactly the memory
          # layout we need and since the file is memory mapped, the
          # memory consumption only happens once per worker for all
          # python processes
          model = SARModel(cache_path_input)
          preds = model.predict(
              df["idx"].values, df["rating"].values, top_k, remove_seen
          )
  
          user = df[local_header["col_user"]].iloc[0]
  
          preds_ret = pd.DataFrame(
              [(user, x.id, x.score) for x in preds], columns=range(3)
          )
  
          return preds_ret
  
      log.info("sarplus.recommend_k_items 3/3: compute recommendations")
  
      df_preds = (
          pred_input.repartition(
              n_user_prediction_partitions, self.header["col_user"]
          )
          .groupby(self.header["col_user"])
          .apply(sar_predict_udf)
      )
  
      df_preds.createOrReplaceTempView(
          self._format("{prefix}predictions")
      )
  
      query = self._format(
          """
              SELECT userID AS `{col_user}`, b.i1 AS `{col_item}`, score
              FROM `{prefix}predictions` AS p, `{prefix}item_mapping` AS b
              WHERE p.itemID = b.idx
          """
      )
      return self.spark.sql(query)
  
file_path: recommenders/contrib/sarplus/python/pysarplus/SARPlus.py
imports: import logging
  import pandas
  from pyspark.sql.types import StructType, StructField, IntegerType, FloatType
  from pyspark.sql.functions import pandas_udf, PandasUDFType
  from pysarplus import SARModel
  import pyspark.sql.functions
label_desc: Use image node_id 5 for calling the SARPlus obj's underlying member method code with example usage: obj._recommend_k_items_fast(test, top_k, remove_seen, n_user_prediction_partitions) and returns: self, preds_ret