Spark Collaborative Filtering RDD中将用户和项目特征保存到HDFS

Question

我想在 Spark 中使用 ALS 从协同过滤的结果中提取用户和项目特征（潜在因素）。到目前为止我的代码：

import org.apache.spark.mllib.recommendation.ALS
import org.apache.spark.mllib.recommendation.MatrixFactorizationModel
import org.apache.spark.mllib.recommendation.Rating

// Load and parse the data
val data = sc.textFile("myhdfs/inputdirectory/als.data")
val ratings = data.map(_.split(',') match { case Array(user, item, rate) =>
  Rating(user.toInt, item.toInt, rate.toDouble)
})

// Build the recommendation model using ALS
val rank = 10
val numIterations = 10
val model = ALS.train(ratings, rank, numIterations, 0.01)

// extract users latent factors
val users = model.userFeatures

// extract items latent factors
val items = model.productFeatures

// save to HDFS
users.saveAsTextFile("myhdfs/outputdirectory/users") // does not work as expected
items.saveAsTextFile("myhdfs/outputdirectory/items") // does not work as expected

但是，写入 HDFS 的内容并不是我所期望的。我希望每一行都有一个元组 (userId, Array_of_doubles)。相反，我看到以下内容：

[myname@host dir]$ hadoop fs -cat myhdfs/outputdirectory/users/*
(1,[D@3c3137b5)
(3,[D@505d9755)
(4,[D@241a409a)
(2,[D@c8c56dd)
.
.

它正在转储数组的哈希值而不是整个数组。我对 print 所需的值执行了以下操作：

for (user <- users) {
  val (userId, lf) = user
  val str = "user:" + userId + "\t" + lf.mkString(" ")
  println(str)
}

这确实打印了我想要的内容，但我无法写入 HDFS（这在控制台上打印）。

我应该怎么做才能将完整的数组正确写入 HDFS？

Spark 版本为 1.2.1。

Answer 1

@JohnTitusJungao 是对的，以下几行也按预期工作：

users.saveAsTextFile("myhdfs/outputdirectory/users") 
items.saveAsTextFile("myhdfs/outputdirectory/items")

这就是原因，userFeatures returns 和 RDD[(Int,Array[Double])]。数组值由您在输出中看到的符号表示，例如[D@3c3137b5 ， D 表示双精度，后跟 @ 和使用 Java toString 方法为此类对象创建的十六进制代码。更多关于 .

val users: RDD[(Int, Array[Double])] = model.userFeatures

要解决这个问题，您需要将数组设为字符串：

val users: RDD[(Int, String)] = model.userFeatures.mapValues(_.mkString(","))

物品也是如此。

Spark Collaborative Filtering RDD中将用户和项目特征保存到HDFS

Saving users and items features to HDFS in Spark Collaborative filtering RDD

arrays

hdfs

apache-spark

rdd