class_docstr: SAR implementation for PySpark.
docstr: Main fit method for SAR.
  
  Expects the dataframes to have row_id, col_id columns which
  are indexes, i.e. contain the sequential integer index of the
  original alphanumeric user and item IDs.  Dataframe also
  contains rating and timestamp as floats; timestamp is in
  seconds since Epoch by default.
  
  Arguments:
      df (pySpark.DataFrame): input dataframe which contains the
          index of users and items.
docstr_tok: ["Main","fit","method","for","SAR",".","Expects","the","dataframes","to","have","row_id",",","col_id","columns","which","are","indexes",",","i.e",".","contain","the","sequential","integer","index","of","the","original","alphanumeric","user","and","item","IDs",".","Dataframe","also","contains","rating","and","timestamp","as","floats",";","timestamp","is","in","seconds","since","Epoch","by","default",".","Arguments",":","df","(","pySpark.DataFrame",")",":","input","dataframe","which","contains","the","index","of","users","and","items","."]
code: def fit(self, df):
      """Main fit method for SAR.
  
      Expects the dataframes to have row_id, col_id columns which
      are indexes, i.e. contain the sequential integer index of the
      original alphanumeric user and item IDs.  Dataframe also
      contains rating and timestamp as floats; timestamp is in
      seconds since Epoch by default.
  
      Arguments:
          df (pySpark.DataFrame): input dataframe which contains the
              index of users and items.
      """
  
      df.createOrReplaceTempView(self._format("{prefix}df_train_input"))
  
      if self.timedecay_formula:
          # With time decay, we compute a sum over ratings given by
          # a user in the case when T=np.inf, so user gets a
          # cumulative sum of ratings for a particular item and not
          # the last rating.  Time Decay does a group by on user
          # item pairs and apply the formula for time decay there
          # Time T parameter is in days and input time is in
          # seconds, so we do dt/60/(T*24*60)=dt/(T*24*3600) the
          # following is the query which we want to run
  
          if self.header["time_now"] is None:
              query = self._format(
                  """
                      SELECT CAST(MAX(`{col_timestamp}`) AS long)
                      FROM `{prefix}df_train_input`
                  """
              )
              self.header["time_now"] = self.spark.sql(query).first()[0]
  
          query = self._format(
              """
                  SELECT `{col_user}`,
                         `{col_item}`,
                         SUM(
                             `{col_rating}` *
                             POW(2, (CAST(`{col_timestamp}` AS LONG) - {time_now}) / {time_decay_half_life})
                            ) AS `{col_rating}`
                  FROM `{prefix}df_train_input`
                  GROUP BY `{col_user}`, `{col_item}`
                  CLUSTER BY `{col_user}`
              """
          )
  
          # replace with time-decayed version
          df = self.spark.sql(query)
      else:
          if self.header["col_timestamp"] in df.columns:
              # we need to de-duplicate items by using the latest item
              query = self._format(
                  """
                      SELECT `{col_user}`, `{col_item}`, `{col_rating}`
                      FROM (
                            SELECT `{col_user}`,
                                   `{col_item}`,
                                   `{col_rating}`,
                                   ROW_NUMBER() OVER user_item_win AS latest
                            FROM `{prefix}df_train_input`
                           ) AS reverse_chrono_table
                      WHERE reverse_chrono_table.latest = 1
                      WINDOW user_item_win AS (
                          PARTITION BY `{col_user}`,`{col_item}`
                          ORDER BY `{col_timestamp}` DESC)
                  """
              )
  
              df = self.spark.sql(query)
  
      df.createOrReplaceTempView(self._format("{prefix}df_train"))
  
      log.info("sarplus.fit 1/2: compute item cooccurrences...")
  
      # compute cooccurrence above minimum threshold
      query = self._format(
          """
              SELECT a.`{col_item}` AS i1,
                     b.`{col_item}` AS i2,
                     COUNT(*) AS value
              FROM `{prefix}df_train` AS a
              INNER JOIN `{prefix}df_train` AS b
              ON a.`{col_user}` = b.`{col_user}` AND a.`{col_item}` <= b.`{col_item}`
              GROUP BY i1, i2
              HAVING value >= {threshold}
              CLUSTER BY i1, i2
          """
      )
  
      item_cooccurrence = self.spark.sql(query)
      item_cooccurrence.write.mode("overwrite").saveAsTable(
          self._format("{prefix}item_cooccurrence")
      )
  
      # compute item frequencies
      self.item_frequencies = item_cooccurrence.filter(
          F.col("i1") == F.col("i2")
      ).select(
          F.col("i1").alias("item_id"),
          F.col("value").alias("frequency"),
      )
  
      # compute the diagonal used later for Jaccard and Lift
      if (
          self.similarity_type == SIM_LIFT
          or self.similarity_type == SIM_JACCARD
      ):
          query = self._format(
              """
                  SELECT i1 AS i, value AS margin
                  FROM `{prefix}item_cooccurrence`
                  WHERE i1 = i2
              """
          )
          item_marginal = self.spark.sql(query)
          item_marginal.createOrReplaceTempView(
              self._format("{prefix}item_marginal")
          )
  
      if self.similarity_type == SIM_COOCCUR:
          self.item_similarity = item_cooccurrence
      elif self.similarity_type == SIM_JACCARD:
          query = self._format(
              """
                  SELECT i1, i2, value / (m1.margin + m2.margin - value) AS value
                  FROM `{prefix}item_cooccurrence` AS a
                  INNER JOIN `{prefix}item_marginal` AS m1 ON a.i1 = m1.i
                  INNER JOIN `{prefix}item_marginal` AS m2 ON a.i2 = m2.i
                  CLUSTER BY i1, i2
              """
          )
          self.item_similarity = self.spark.sql(query)
      elif self.similarity_type == SIM_LIFT:
          query = self._format(
              """
                  SELECT i1, i2, value / (m1.margin * m2.margin) AS value
                  FROM `{prefix}item_cooccurrence` AS a
                  INNER JOIN `{prefix}item_marginal` AS m1 ON a.i1 = m1.i
                  INNER JOIN `{prefix}item_marginal` AS m2 ON a.i2 = m2.i
                  CLUSTER BY i1, i2
              """
          )
          self.item_similarity = self.spark.sql(query)
      else:
          raise ValueError(
              "Unknown similarity type: {0}".format(
                  self.similarity_type
              )
          )
  
      # store upper triangular
      log.info(
          "sarplus.fit 2/2: compute similarity metric %s..."
          % self.similarity_type
      )
      self.item_similarity.write.mode("overwrite").saveAsTable(
          self._format("{prefix}item_similarity_upper")
      )
  
      # expand upper triangular to full matrix
  
      query = self._format(
          """
              SELECT i1, i2, value
              FROM (
                    (
                     SELECT i1, i2, value
                     FROM `{prefix}item_similarity_upper`
                    )
                    UNION ALL
                    (
                     SELECT i2 AS i1, i1 AS i2, value
                     FROM `{prefix}item_similarity_upper`
                     WHERE i1 <> i2
                    )
                   )
              CLUSTER BY i1
          """
      )
  
      self.item_similarity = self.spark.sql(query)
      self.item_similarity.write.mode("overwrite").saveAsTable(
          self._format("{prefix}item_similarity")
      )
  
      # free space
      self.spark.sql(
          self._format("DROP TABLE `{prefix}item_cooccurrence`")
      )
      self.spark.sql(
          self._format("DROP TABLE `{prefix}item_similarity_upper`")
      )
  
      self.item_similarity = self.spark.table(
          self._format("{prefix}item_similarity")
      )
  
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 3 for calling the SARPlus obj's underlying member method code with example usage: obj.fit(df) without return types