我已经成功地在以下格式的平面文件上运行了Mahout rowsimilation:
物品id标签1标签2标签3
这必须通过cli运行,并且输出也是平面文件。我想让它从MongoDB读取数据(也可以使用其他DB),然后将输出转储到DB,然后可以从我们的系统中提取。
我研究了过去几天,发现了以下内容:
- 必须编写实现RowSimilarity的Scala代码
- 将IndexedDataSet对象传递给它以处理数据
- 将输出转换为所需格式(json.csv)
我还没有弄清楚如何将数据从DB导入IndexedDataSet。此外,我已经阅读了RDD格式,但仍然不知道如何将json数据转换为RDD,以便RowSimilarity代码使用。
tl;dr:如何转换MongoDB数据,以便通过mahout/spark-rowsimilation进行处理
第1版:我从这个链接找到了一些将Mongodate转换为RDD的代码:https://github.com/mongodb/mongo-hadoop/wiki/Spark-Usage#scala-示例
现在我需要帮助将其转换为IndexedDataset,以便将其传递给SimilarityAnalysis.rowSimilarityIDS.
tl;dr:如何将RDD转换为IndexedDataset 以下是答案: build.sbt: 我使用mongo-hadoop从mongo获取数据并使用它。由于我的数据有一个数组,我不得不使用flatMapValues将其压平,然后传递给IDS以获得正确的输出。 附言:我在这里发布了答案,而不是链接的问题,因为这个问答;A涵盖了获取数据和处理数据的全部范围。import org.apache.hadoop.conf.Configuration
import org.apache.mahout.math.cf.SimilarityAnalysis
import org.apache.mahout.math.indexeddataset.Schema
import org.apache.mahout.sparkbindings
import org.apache.mahout.sparkbindings.indexeddataset.IndexedDatasetSpark
import org.apache.spark.rdd.RDD
import org.bson.BSONObject
import com.mongodb.hadoop.MongoInputFormat
object SparkExample extends App {
implicit val mc = sparkbindings.mahoutSparkContext(masterUrl = "local", appName = "RowSimilarity")
val mongoConfig = new Configuration()
mongoConfig.set("mongo.input.uri", "mongodb://hostname:27017/db.collection")
val documents: RDD[(Object, BSONObject)] = mc.newAPIHadoopRDD(
mongoConfig,
classOf[MongoInputFormat],
classOf[Object],
classOf[BSONObject]
)
val documents_Array: RDD[(String, Array[String])] = documents.map(
doc1 => (
doc1._2.get("product_id").toString(),
doc1._2.get("product_attribute_value").toString().replace("[ "", "").replace(""]", "").split("" , "").map(value => value.toLowerCase.replace(" ", "-").mkString(" "))
)
)
val new_doc: RDD[(String, String)] = documents_Array.flatMapValues(x => x)
val myIDs = IndexedDatasetSpark(new_doc)(mc)
val readWriteSchema = new Schema(
"rowKeyDelim" -> "t",
"columnIdStrengthDelim" -> ":",
"omitScore" -> false,
"elementDelim" -> " "
)
SimilarityAnalysis.rowSimilarityIDS(myIDs).dfsWrite("hdfs://hadoop:9000/mongo-hadoop-rowsimilarity", readWriteSchema)(mc)
}
name := "scala-mongo"
version := "1.0"
scalaVersion := "2.10.6"
libraryDependencies += "org.mongodb" %% "casbah" % "3.1.1"
libraryDependencies += "org.apache.spark" %% "spark-core" % "1.6.1"
libraryDependencies += "org.mongodb.mongo-hadoop" % "mongo-hadoop-core" % "1.4.2"
libraryDependencies ++= Seq(
"org.apache.hadoop" % "hadoop-client" % "2.6.0" exclude("javax.servlet", "servlet-api") exclude ("com.sun.jmx", "jmxri") exclude ("com.sun.jdmk", "jmxtools") exclude ("javax.jms", "jms") exclude ("org.slf4j", "slf4j-log4j12") exclude("hsqldb","hsqldb"),
"org.scalatest" % "scalatest_2.10" % "1.9.2" % "test"
)
libraryDependencies += "org.apache.mahout" % "mahout-math-scala_2.10" % "0.11.2"
libraryDependencies += "org.apache.mahout" % "mahout-spark_2.10" % "0.11.2"
libraryDependencies += "org.apache.mahout" % "mahout-math" % "0.11.2"
libraryDependencies += "org.apache.mahout" % "mahout-hdfs" % "0.11.2"
resolvers += "typesafe repo" at " http://repo.typesafe.com/typesafe/releases/"
resolvers += Resolver.mavenLocal