Как использовать mllib.recommendation, если идентификаторы пользователя являются строкой, а не смежными целыми числами?

Ответ 1

Spark действительно не требует числового id, он просто должен иметь некоторое уникальное значение, но для реализации они выбрали Int.

Вы можете сделать простые преобразования назад и вперед для userId:

  case class MyRating(userId: String, product: Int, rating: Double)

  val data: RDD[MyRating] = ???

  // Assign unique Long id for each userId
  val userIdToInt: RDD[(String, Long)] = 
    data.map(_.userId).distinct().zipWithUniqueId()

  // Reverse mapping from generated id to original
  val reverseMapping: RDD[(Long, String)]
    userIdToInt map { case (l, r) => (r, l) }

  // Depends on data size, maybe too big to keep
  // on single machine
  val map: Map[String, Int] = 
    userIdToInt.collect().toMap.mapValues(_.toInt)

  // Transform to MLLib rating
  val rating: RDD[Rating] = data.map { r =>
    Rating(userIdToInt.lookup(r.userId).head.toInt, r.product, r.rating)
    // -- or
    Rating(map(r.userId), r.product, r.rating)
  }

  // ... train model

  // ... get back to MyRating userId from Int

  val someUserId: String = reverseMapping.lookup(123).head

Вы также можете попробовать "data.zipWithUniqueId()", но я не уверен, что в этом случае .toInt будет безопасным преобразованием, даже если размер набора данных невелик.

Ответ 2

Вам нужно запустить StringIndexer по вашим идентификаторам пользователей, чтобы преобразовать строку в уникальный целочисленный индекс. Они не должны быть непрерывными.

Мы используем это для нашего механизма рекомендации элемента в https://www.aihello.com

df (пользователь: String, product, rating)

  val stringindexer = new StringIndexer()
      .setInputCol("user")
      .setOutputCol("userNumber")
  val modelc = stringindexer.fit(df)
  val  df = modelc.transform(df)

Ответ 3

Вышеупомянутое решение может не всегда работать, как я обнаружил. Spark не может выполнять преобразования RDD из других RDD. Выход ошибки:

org.apache.spark.SparkException: преобразования и действия RDD могут только введите код, указанный водителем, а не внутри другого преобразования; например, rdd1.map(x = > rdd2.values.count() * x) является недопустимым, поскольку преобразование значений и действие счета не могут выполняться внутри преобразования rdd1.map. Для большего информацию см. в SPARK-5063.

В качестве решения вы можете присоединиться к userIdToInt RDD с исходными данными RDD, чтобы сохранить связь между userId и uniqueId. Затем вы можете снова присоединиться к результатам RDD с этим RDD.

// Create RDD with the unique id included
val dataWithUniqueUserId: RDD[(String, Int, Int, Double)] = 
    data.keyBy(_.userId).join(userIdToInt).map(r => 
        (r._2._1.userId, r._2._2.toInt, r._2._1.productId, 1))

Ответ 4

@Ganesh Кришнан прав, StringIndexer решает эту проблему.

from pyspark.ml.feature import OneHotEncoder, StringIndexer
from pyspark.sql import SQLContext
>>> spark = SQLContext(sc)                                                                             
>>> df = spark.createDataFrame(
...     [(0, "a"), (1, "b"), (2, "c"), (3, "a"), (4, "a"), (5, "c")],
...     ["id", "category"])

| id|category|
+---+--------+
|  0|       a|
|  1|       b|
|  2|       c|
|  3|       a|
|  4|       a|
|  5|       c|
+---+--------+
>>> stringIndexer = StringIndexer(inputCol="category", outputCol="categoryIndex")
>>> model = stringIndexer.fit(df)
>>> indexed = model.transform(df)
>>> indexed.show()
+---+--------+-------------+
| id|category|categoryIndex|
+---+--------+-------------+
|  0|       a|          0.0|
|  1|       b|          2.0|
|  2|       c|          1.0|
|  3|       a|          0.0|
|  4|       a|          0.0|
|  5|       c|          1.0|
+---+--------+-------------+

>>> converter = IndexToString(inputCol="categoryIndex", outputCol="originalCategory")
>>> converted = converter.transform(indexed)
>>> converted.show()
+---+--------+-------------+----------------+
| id|category|categoryIndex|originalCategory|
+---+--------+-------------+----------------+
|  0|       a|          0.0|               a|
|  1|       b|          2.0|               b|
|  2|       c|          1.0|               c|
|  3|       a|          0.0|               a|
|  4|       a|          0.0|               a|
|  5|       c|          1.0|               c|
+---+--------+-------------+----------------+

>>> converted.select("id", "originalCategory").show()
+---+----------------+
| id|originalCategory|
+---+----------------+
|  0|               a|
|  1|               b|
|  2|               c|
|  3|               a|
|  4|               a|
|  5|               c|
+---+----------------+