Извлечение значений из Scala WrappedArray

Question

Я работаю с моделью ALS Apache Spark, а метод рекомендуемаяForAllUsers возвращает фрейм данных со схемой

root
 |-- user_id: integer (nullable = false)
 |-- recommendations: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- item_id: integer (nullable = true)
 |    |    |-- rating: float (nullable = true)

На практике рекомендации имеют вид WrappedArray:

WrappedArray([636958,0.32910484], [995322,0.31974298], [1102140,0.30444127], [1160820,0.27908015], [1208899,0.26943958])

Я пытаюсь извлечь просто item_ids и вернуть их как одномерный массив. Таким образом, приведенный выше пример будет [636958,995322,1102140,1160820,1208899]

Это то, что доставляет мне неприятности. Пока что у меня есть:

    val numberOfRecs = 20
    val userRecs = model.recommendForAllUsers(numberOfRecs).cache()

    val strippedScores = userRecs.rdd.map(row => {
      val user_id = row.getInt(0)
      val recs = row.getAs[Seq[Row]](1)

      val item_ids = new Array[Int](numberOfRecs)

      recs.toArray.foreach(x => {
        item_ids :+ x.get(0)
      })

      item_ids
    })

Но это просто возвращает [I@2f318251, и если я получу строковое значение через mkString (","), он вернет 0,0,0,0,0,0

Есть мысли о том, как я могу извлечь item_ids и вернуть их в виде отдельного одномерного массива?

Answer 1

В документах Spark ALSModel обнаружено, что recommendForAllUsers возвращает

"DataFrame из (userCol: Int, рекомендации), где рекомендации хранятся в виде массива (itemCol: Int, rating:Float) Rows "(https://spark.apache.org/docs/2.2.0/api/scala/index.html#org.apache.spark.ml.recommendation.ALSModel)

По массиву это означает WrappedArray, поэтому вместо того, чтобы пытаться привести его к Seq[Row], я приведу его к mutable.WrappedArray[Row]. Затем я смогполучить каждый item_id, например:

    val userRecItems = userRecs.rdd.map(row => {
      val user_id = row.getInt(0)
      val recs = row.getAs[mutable.WrappedArray[Row]](1)

      for (rec <- recs) {
        val item_id = rec.getInt(0)
        userRecommendatinos += game_id
      }
    })

, где userRecommendations был изменяемым ArrayBuffer

Answer 2

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

    scala> case class Recommendation(item_id: Int, rating: Float)
defined class Recommendation

scala> val userReqs = Seq(Array(Recommendation(636958,0.32910484f), Recommendation(995322,0.31974298f), Recommendation(1102140,0.30444127f), Recommendation(1160820,0.27908015f), Recommendation(1208899,0.26943958f))).toDF
userReqs: org.apache.spark.sql.DataFrame = [value: array<struct<item_id:int,rating:float>>]

scala> userReqs.printSchema
root
 |-- value: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- item_id: integer (nullable = false)
 |    |    |-- rating: float (nullable = false)


scala> userReqs.select("value.item_id").show(false)
+-------------------------------------------+
|item_id                                    |
+-------------------------------------------+
|[636958, 995322, 1102140, 1160820, 1208899]|
+-------------------------------------------+

scala> val ids = userReqs.select("value.item_id").collect().flatMap(_.getAs[Seq[Int]](0))
ids: Array[Int] = Array(636958, 995322, 1102140, 1160820, 1208899)