通过 pyspark.ml CrossValidator 调整隐式 pyspark.ml ALS 矩阵分解模型的参数

我正在尝试调整使用隐式数据的 ALS 矩阵分解模型的参数。为此，我尝试使用 pyspark.ml.tuning.CrossValidator 运行参数网格并选择最佳模型。我相信我的问题出在评估器上，但我无法弄清楚。

我可以使用回归 RMSE 求值器将其用于显式数据模型，如下所示：

from pyspark import SparkConf, SparkContext
from pyspark.sql import SQLContext
from pyspark.ml.recommendation import ALS
from pyspark.ml.tuning import CrossValidator, ParamGridBuilder
from pyspark.ml.evaluation import BinaryClassificationEvaluator
from pyspark.ml.evaluation import RegressionEvaluator

from pyspark.sql.functions import rand


conf = SparkConf() \
  .setAppName("MovieLensALS") \
  .set("spark.executor.memory", "2g")
sc = SparkContext(conf=conf)

sqlContext = SQLContext(sc)

dfRatings = sqlContext.createDataFrame([(0, 0, 4.0), (0, 1, 2.0), (1, 1, 3.0), (1, 2, 4.0), (2, 1, 1.0), (2, 2, 5.0)],
                                 ["user", "item", "rating"])
dfRatingsTest = sqlContext.createDataFrame([(0, 0), (0, 1), (1, 1), (1, 2), (2, 1), (2, 2)], ["user", "item"])

alsExplicit = ALS()
defaultModel = alsExplicit.fit(dfRatings)

paramMapExplicit = ParamGridBuilder() \
                    .addGrid(alsExplicit.rank, [8, 12]) \
                    .addGrid(alsExplicit.maxIter, [10, 15]) \
                    .addGrid(alsExplicit.regParam, [1.0, 10.0]) \
                    .build()

evaluatorR = RegressionEvaluator(metricName="rmse", labelCol="rating")

cvExplicit = CrossValidator(estimator=alsExplicit, estimatorParamMaps=paramMapExplicit, evaluator=evaluatorR)
cvModelExplicit = cvExplicit.fit(dfRatings)

predsExplicit = cvModelExplicit.bestModel.transform(dfRatingsTest)
predsExplicit.show()

当我尝试对隐式数据（假设是观看次数而不是评级）执行此操作时，我收到一个我无法完全弄清楚的错误。这是代码（与上面非常相似）：

dfCounts = sqlContext.createDataFrame([(0,0,0), (0,1,12), (0,2,3), (1,0,5), (1,1,9), (1,2,0), (2,0,0), (2,1,11), (2,2,25)],
                                 ["user", "item", "rating"])
dfCountsTest = sqlContext.createDataFrame([(0, 0), (0, 1), (1, 1), (1, 2), (2, 1), (2, 2)], ["user", "item"])

alsImplicit = ALS(implicitPrefs=True)
defaultModelImplicit = alsImplicit.fit(dfCounts)

paramMapImplicit = ParamGridBuilder() \
                    .addGrid(alsImplicit.rank, [8, 12]) \
                    .addGrid(alsImplicit.maxIter, [10, 15]) \
                    .addGrid(alsImplicit.regParam, [1.0, 10.0]) \
                    .addGrid(alsImplicit.alpha, [2.0,3.0]) \
                    .build()

evaluatorB = BinaryClassificationEvaluator(metricName="areaUnderROC", labelCol="rating")
evaluatorR = RegressionEvaluator(metricName="rmse", labelCol="rating")

cv = CrossValidator(estimator=alsImplicit, estimatorParamMaps=paramMapImplicit, evaluator=evaluatorR)
cvModel = cv.fit(dfCounts)

predsImplicit = cvModel.bestModel.transform(dfCountsTest)
predsImplicit.show()

我尝试使用 RMSE 评估器执行此操作，但出现错误。据我了解，我还应该能够将 AUC 度量用于二​​元分类评估器，因为隐式矩阵分解的预测是用于二元矩阵 p_ui 预测的置信矩阵 c_ui根据本文 http://yifanhu.net/PUB/cf.pdf，pyspark ALS 的文档引用了该内容。

使用任一评估器都会出现错误，而且我在网上找不到任何关于交叉验证隐式 ALS 模型的富有成效的讨论。我正在查看 CrossValidator 源代码，试图找出问题所在，但遇到了麻烦。我的想法之一是，在该过程将隐式数据矩阵 r_ui 转换为二进制矩阵 p_ui 和置信矩阵 c_ui 后，我不确定在评估阶段将预测的 c_ui 矩阵与什么进行比较。

这是错误：

Traceback (most recent call last):

  File "<ipython-input-16-6c43b997005e>", line 1, in <module>
    cvModel = cv.fit(dfCounts)

  File "C:/spark-1.6.1-bin-hadoop2.6/python\pyspark\ml\pipeline.py", line 69, in fit
    return self._fit(dataset)

  File "C:/spark-1.6.1-bin-hadoop2.6/python\pyspark\ml\tuning.py", line 239, in _fit
    model = est.fit(train, epm[j])

  File "C:/spark-1.6.1-bin-hadoop2.6/python\pyspark\ml\pipeline.py", line 67, in fit
    return self.copy(params)._fit(dataset)

  File "C:/spark-1.6.1-bin-hadoop2.6/python\pyspark\ml\wrapper.py", line 133, in _fit
    java_model = self._fit_java(dataset)

  File "C:/spark-1.6.1-bin-hadoop2.6/python\pyspark\ml\wrapper.py", line 130, in _fit_java
    return self._java_obj.fit(dataset._jdf)

  File "C:\spark-1.6.1-bin-hadoop2.6\python\lib\py4j-0.9-src.zip\py4j\java_gateway.py", line 813, in __call__
    answer, self.gateway_client, self.target_id, self.name)

  File "C:/spark-1.6.1-bin-hadoop2.6/python\pyspark\sql\utils.py", line 45, in deco
    return f(*a, **kw)

  File "C:\spark-1.6.1-bin-hadoop2.6\python\lib\py4j-0.9-src.zip\py4j\protocol.py", line 308, in get_return_value
    format(target_id, ".", name), value)

etc.......

UPDATE

我尝试缩放输入，使其在 0 到 1 的范围内，并使用 RMSE 求值器。在我尝试将其插入 CrossValidator 之前，它似乎运行良好。

以下代码有效。我得到预测，并从评估员那里得到 RMSE 值。

from pyspark import SparkConf, SparkContext
from pyspark.sql import SQLContext
from pyspark.sql.types import FloatType
import pyspark.sql.functions as F
from pyspark.ml.recommendation import ALS
from pyspark.ml.tuning import CrossValidator, ParamGridBuilder
from pyspark.ml.evaluation import RegressionEvaluator


conf = SparkConf() \
  .setAppName("ALSPractice") \
  .set("spark.executor.memory", "2g")
sc = SparkContext(conf=conf)

sqlContext = SQLContext(sc)

# Users 0, 1, 2, 3 - Items 0, 1, 2, 3, 4, 5 - Ratings 0.0-5.0
dfCounts2 = sqlContext.createDataFrame([(0,0,5.0), (0,1,5.0),            (0,3,0.0), (0,4,0.0), 
                                        (1,0,5.0),            (1,2,4.0), (1,3,0.0), (1,4,0.0),
                                        (2,0,0.0),            (2,2,0.0), (2,3,5.0), (2,4,5.0),
                                        (3,0,0.0), (3,1,0.0),            (3,3,4.0)            ],
                                       ["user", "item", "rating"])

dfCountsTest2 = sqlContext.createDataFrame([(0,0), (0,1), (0,2), (0,3), (0,4),
                                            (1,0), (1,1), (1,2), (1,3), (1,4),
                                            (2,0), (2,1), (2,2), (2,3), (2,4),
                                            (3,0), (3,1), (3,2), (3,3), (3,4)], ["user", "item"])

# Normalize rating data to [0,1] range based on max rating
colmax = dfCounts2.select(F.max('rating')).collect()[0].asDict().values()[0]
normalize = udf(lambda x: x/colmax, FloatType())
dfCountsNorm = dfCounts2.withColumn('ratingNorm', normalize(col('rating')))

alsImplicit = ALS(implicitPrefs=True)
defaultModelImplicit = alsImplicit.fit(dfCountsNorm)
preds = defaultModelImplicit.transform(dfCountsTest2)

evaluatorR2 = RegressionEvaluator(metricName="rmse", labelCol="ratingNorm")
evaluatorR2.evaluate(defaultModelImplicit.transform(dfCountsNorm))

preds = defaultModelImplicit.transform(dfCountsTest2)

我不明白的是为什么以下不起作用。我使用相同的估计器、相同的评估器并拟合相同的数据。为什么这些可以在上面工作但不能在 CrossValidator 中工作：

paramMapImplicit = ParamGridBuilder() \
                    .addGrid(alsImplicit.rank, [8, 12]) \
                    .addGrid(alsImplicit.maxIter, [10, 15]) \
                    .addGrid(alsImplicit.regParam, [1.0, 10.0]) \
                    .addGrid(alsImplicit.alpha, [2.0,3.0]) \
                    .build()

cv = CrossValidator(estimator=alsImplicit, estimatorParamMaps=paramMapImplicit, evaluator=evaluatorR2)
cvModel = cv.fit(dfCountsNorm)

忽略技术问题，严格来说，考虑到 ALS 生成的带有隐式反馈的输入，这两种方法都不正确。

• 你不能使用RegressionEvaluator因为，正如您所知，预测可以解释为置信度值，并表示为 [0, 1] 范围内的浮点数，而标签列只是一个未绑定的整数。这些值显然没有可比性。
• 你不能使用BinaryClassificationEvaluator因为即使预测可以解释为概率标签也不代表二元决策。此外，预测列的类型无效，无法直接使用BinaryClassificationEvaluator

您可以尝试转换其中一列，以便输入满足要求，但这从理论角度来看并不是真正合理的方法，并且引入了难以调整的附加参数。

• 将标签列映射到 [0, 1] 范围并使用 RMSE。

• 将标签列转换为具有固定阈值和扩展的二进制指示器ALS / ALSModel返回预期的列类型。假设阈值是1，它可能是这样的

  from pyspark.ml.recommendation import *
  from pyspark.sql.functions import udf, col
  from pyspark.mllib.linalg import DenseVector, VectorUDT
  
  class BinaryALS(ALS):
      def fit(self, df):
          assert self.getImplicitPrefs()
          model = super(BinaryALS, self).fit(df)
          return ALSBinaryModel(model._java_obj)
  
  class ALSBinaryModel(ALSModel):
      def transform(self, df):
          transformed = super(ALSBinaryModel, self).transform(df)
          as_vector = udf(lambda x: DenseVector([1 - x, x]), VectorUDT())
          return transformed.withColumn(
              "rawPrediction", as_vector(col("prediction")))
  
  # Add binary label column
  with_binary = dfCounts.withColumn(
      "label_binary", (col("rating") > 0).cast("double"))
  
  als_binary_model = BinaryALS(implicitPrefs=True).fit(with_binary)
  
  evaluatorB = BinaryClassificationEvaluator(
      metricName="areaUnderROC", labelCol="label_binary")
  
  evaluatorB.evaluate(als_binary_model.transform(with_binary))
  ## 1.0
  

一般来说，教科书中缺少有关使用隐式反馈评估推荐系统的材料，我建议您阅读eliasah https://stackoverflow.com/users/3415409/eliasah's answer https://stackoverflow.com/questions/46462470/how-can-i-evaluate-the-implicit-feedback-als-algorithm-for-recommendations-in-ap/46490352#46490352关于评估此类推荐人。