python - Parámetros de ajuste para el modelo de factorización de matriz ALS pyspark.ml implícito a través de pyspark.ml CrossValidator

apache-spark apache-spark-ml

Question

Estoy tratando de ajustar los parámetros de un modelo de factorización de matriz ALS que usa datos implícitos. Para esto, estoy tratando de usar pyspark.ml.tuning.CrossValidator para ejecutar una cuadrícula de parámetros y seleccionar el mejor modelo. Creo que mi problema está en el evaluador, pero no puedo resolverlo.

Puedo hacer que esto funcione para un modelo de datos explícito con un evaluador de regresión RMSE, de la siguiente manera:

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()

Cuando trato de hacer esto para datos implícitos (digamos recuentos de vistas en lugar de calificaciones), recibo un error que no puedo entender. Aquí está el código (muy similar al anterior):

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()

Intenté hacer esto con un evaluador RMSE y recibí un error. Según tengo entendido, también debería poder usar la métrica de AUC para el evaluador de clasificación binaria, porque las predicciones de la factorización de matriz implícita son una matriz de confianza c_ui para las predicciones de una matriz binaria p_ui según este documento , que la documentación para pyspark ALS cita.

El uso de cualquiera de los evaluadores me da un error y no puedo encontrar ninguna discusión fructífera sobre la validación cruzada de modelos ALS implícitos en línea. Estoy revisando el código fuente de CrossValidator para tratar de descubrir qué está mal, pero tengo problemas. Una de mis ideas es que después de que el proceso convierte la matriz de datos implícita r_ui en la matriz binaria p_ui y la matriz de confianza c_ui, no estoy seguro de qué compara la matriz c_ui predicha durante la etapa de evaluación.

Aquí está el error:

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.......

ACTUALIZAR

Intenté escalar la entrada para que esté en el rango de 0 a 1 y use un evaluador RMSE. Parece funcionar bien hasta que trato de insertarlo en el CrossValidator.

El siguiente código funciona. Recibo predicciones y obtengo un valor RMSE de mi evaluador.

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)

Lo que no entiendo es por qué lo siguiente no funciona. Estoy usando el mismo estimador, el mismo evaluador y ajustando los mismos datos. ¿Por qué estos funcionarían arriba pero no dentro del 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)

Answer 1

Con los comentarios implícitos no tenemos reacciones de los usuarios a nuestras recomendaciones. Por lo tanto, no podemos usar métricas basadas en precisión.

En el artículo ya citado , se utiliza la métrica de clasificación de percentil esperada.

Puede intentar implementar un Evaluador basado en una métrica similar en la biblioteca Spark ML y utilizarlo en su canalización de Validación cruzada.

Answer 2

Ignorando problemas técnicos, estrictamente hablando, ninguno de los métodos es correcto dada la entrada generada por ALS con retroalimentación implícita.

• no puede usar RegressionEvaluator porque, como ya sabe, la predicción se puede interpretar como un valor de confianza y se representa como un número de coma flotante en el rango [0, 1] y la columna de etiqueta es solo un entero independiente. Estos valores claramente no son comparables.
• no puede usar BinaryClassificationEvaluator porque incluso si la predicción puede interpretarse como una etiqueta de probabilidad no representa una decisión binaria. Además, la columna de predicción tiene un tipo no válido y no se puede usar directamente con BinaryClassificationEvaluator

Puede intentar convertir una de las columnas para que la entrada se ajuste a los requisitos, pero este no es realmente un enfoque justificado desde una perspectiva teórica e introduce parámetros adicionales que son difíciles de ajustar.

• mapear la columna de etiqueta al rango [0, 1] y usar RMSE.

• convierta la columna de etiqueta en indicador binario con umbral fijo y extienda ALS / ALSModel para devolver el tipo de columna esperado. Suponiendo que el valor umbral es 1, podría ser algo como esto

  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

En términos generales, el material sobre la evaluación de los sistemas de recomendación con retroalimentación implícita falta en los libros de texto, le sugiero que lea la answer eliasah sobre la evaluación de este tipo de recomendaciones.