class_docstr: Spark Rating Evaluator
class_docstr_tok: ["Spark","Rating","Evaluator"]
docstr: Initializer.
  
  This is the Spark version of rating metrics evaluator.
  The methods of this class, calculate rating metrics such as root mean squared error, mean absolute error,
  R squared, and explained variance.
  
  Args:
      rating_true (pyspark.sql.DataFrame): True labels.
      rating_pred (pyspark.sql.DataFrame): Predicted labels.
      col_user (str): column name for user.
      col_item (str): column name for item.
      col_rating (str): column name for rating.
      col_prediction (str): column name for prediction.
docstr_tok: ["Initializer",".","This","is","the","Spark","version","of","rating","metrics","evaluator",".","The","methods","of","this","class",",","calculate","rating","metrics","such","as","root","mean","squared","error",",","mean","absolute","error",",","R","squared",",","and","explained","variance",".","Args",":","rating_true","(","pyspark.sql.DataFrame",")",":","True","labels",".","rating_pred","(","pyspark.sql.DataFrame",")",":","Predicted","labels",".","col_user","(","str",")",":","column","name","for","user",".","col_item","(","str",")",":","column","name","for","item",".","col_rating","(","str",")",":","column","name","for","rating",".","col_prediction","(","str",")",":","column","name","for","prediction","."]
code: def __init__(
      self,
      rating_true,
      rating_pred,
      col_user=DEFAULT_USER_COL,
      col_item=DEFAULT_ITEM_COL,
      col_rating=DEFAULT_RATING_COL,
      col_prediction=DEFAULT_PREDICTION_COL,
  ):
      """Initializer.
  
      This is the Spark version of rating metrics evaluator.
      The methods of this class, calculate rating metrics such as root mean squared error, mean absolute error,
      R squared, and explained variance.
  
      Args:
          rating_true (pyspark.sql.DataFrame): True labels.
          rating_pred (pyspark.sql.DataFrame): Predicted labels.
          col_user (str): column name for user.
          col_item (str): column name for item.
          col_rating (str): column name for rating.
          col_prediction (str): column name for prediction.
      """
      self.rating_true = rating_true
      self.rating_pred = rating_pred
      self.col_user = col_user
      self.col_item = col_item
      self.col_rating = col_rating
      self.col_prediction = col_prediction
  
      # Check if inputs are Spark DataFrames.
      if not isinstance(self.rating_true, DataFrame):
          raise TypeError(
              "rating_true should be but is not a Spark DataFrame"
          )  # pragma : No Cover
  
      if not isinstance(self.rating_pred, DataFrame):
          raise TypeError(
              "rating_pred should be but is not a Spark DataFrame"
          )  # pragma : No Cover
  
      # Check if columns exist.
      true_columns = self.rating_true.columns
      pred_columns = self.rating_pred.columns
  
      if rating_true.count() == 0:
          raise ValueError("Empty input dataframe")
      if rating_pred.count() == 0:
          raise ValueError("Empty input dataframe")
  
      if self.col_user not in true_columns:
          raise ValueError(
              "Schema of rating_true not valid. Missing User Col"
          )
      if self.col_item not in true_columns:
          raise ValueError(
              "Schema of rating_true not valid. Missing Item Col"
          )
      if self.col_rating not in true_columns:
          raise ValueError(
              "Schema of rating_true not valid. Missing Rating Col"
          )
  
      if self.col_user not in pred_columns:
          raise ValueError(
              "Schema of rating_pred not valid. Missing User Col"
          )  # pragma : No Cover
      if self.col_item not in pred_columns:
          raise ValueError(
              "Schema of rating_pred not valid. Missing Item Col"
          )  # pragma : No Cover
      if self.col_prediction not in pred_columns:
          raise ValueError(
              "Schema of rating_pred not valid. Missing Prediction Col"
          )
  
      self.rating_true = self.rating_true.select(
          col(self.col_user),
          col(self.col_item),
          col(self.col_rating).cast("double").alias("label"),
      )
      self.rating_pred = self.rating_pred.select(
          col(self.col_user),
          col(self.col_item),
          col(self.col_prediction).cast("double").alias("prediction"),
      )
  
      self.y_pred_true = (
          self.rating_true.join(
              self.rating_pred, [self.col_user, self.col_item], "inner"
          )
          .drop(self.col_user)
          .drop(self.col_item)
      )
  
      self.metrics = RegressionMetrics(
          self.y_pred_true.rdd.map(lambda x: (x.prediction, x.label))
      )
  
imports: import numpy
  from recommenders.utils.constants import DEFAULT_PREDICTION_COL, DEFAULT_USER_COL, DEFAULT_ITEM_COL, DEFAULT_RATING_COL, DEFAULT_RELEVANCE_COL, DEFAULT_SIMILARITY_COL, DEFAULT_ITEM_FEATURES_COL, DEFAULT_ITEM_SIM_MEASURE, DEFAULT_TIMESTAMP_COL, DEFAULT_K, DEFAULT_THRESHOLD
label_desc: Use image node_id 1 to create a new SparkRatingEvaluation object  with example: obj = SparkRatingEvaluation(rating_true, rating_pred, col_user, col_item, col_rating, col_prediction)