在火花中为LDA准备数据-Java 学习之路

我正在努力实现Spark LDA模型（通过Scala API），并且我的数据必须采用格式化步骤 . 我的原始数据（存储在文本文件中）采用以下格式，基本上是令牌列表及其对应的文档 . 一个简化的例子：

doc XXXXX   term    XXXXX
1   x       'a'     x
1   x       'a'     x
1   x       'b'     x
2   x       'b'     x
2   x       'd'     x
...

XXXXX列是垃圾数据的地方我不知道我拥有什么 . 我希望从示例中可以清楚地看出，原始数据中每个标记有一行（因此，如果给定的术语在文档中出现5次，则对应于5行文本） .

在任何情况下，我都需要将这些数据格式化为用于运行Spark LDA模型的稀疏项频率向量，但是我不熟悉Scala，因此遇到了一些麻烦 .

我开始：

import org.apache.spark.mllib.clustering.{LDA, DistributedLDAModel}
import org.apache.spark.mllib.linalg.{Vector, Vectors}
import org.apache.spark.rdd.RDD

val corpus:RDD[Array[String]] = sc.textFile("path/to/data")
    .map(_.split('\t')).map(x => Array(x(0),x(2)))

然后我得到生成稀疏向量所需的词汇数据：

val vocab: RDD[String] = corpus.map(_(1)).distinct()
val vocabMap: Map[String, Int] = vocab.collect().zipWithIndex.toMap

我不知道的是在这里使用的正确的映射函数，这样我最终得到每个文档的稀疏项频率向量，然后我可以将其输入到LDA模型中 . 我想我需要这些方面的东西......

val documents: RDD[(Long, Vector)] = corpus.groupBy(_(0)).zipWithIndex
    .map(x =>(x._2,Vectors.sparse(vocabMap.size, ???)))

此时我可以运行实际的LDA：

val lda = new LDA().setK(n_topics)
val ldaModel = lda.run(documents)

基本上，我没有应用于每个组的功能，以便我可以将术语频率数据（可能是 map ？）提供给稀疏矢量 . 换句话说，如何在上面的代码片段中填写 ??? 以达到预期效果？

1 回答

解决这个问题的一种方法：

确保 spark-csv 包可用
将数据加载到DataFrame中并选择感兴趣的列

val df = sqlContext.read
    .format("com.databricks.spark.csv")
    .option("header", "true")
    .option("inferSchema", "true") // Optional, providing schema is prefered
    .option("delimiter", "\t")
    .load("foo.csv")
    .select($"doc".cast("long").alias("doc"), $"term")

index term 专栏：

import org.apache.spark.ml.feature.StringIndexer

val indexer = new StringIndexer()
  .setInputCol("term")
  .setOutputCol("termIndexed")

val indexed = indexer.fit(df)
  .transform(df)
  .drop("term")
  .withColumn("termIndexed", $"termIndexed".cast("integer"))
  .groupBy($"doc", $"termIndexed")
  .agg(count(lit(1)).alias("cnt").cast("double"))

转换为 PairwiseRDD

import org.apache.spark.sql.Row

val pairs = indexed.map{case Row(doc: Long, term: Int, cnt: Double) => 
  (doc, (term, cnt))}

group by doc：

val docs = pairs.groupByKey

创建特征向量

import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.sql.functions.max

val n = indexed.select(max($"termIndexed")).first.getInt(0) + 1

val docsWithFeatures = docs.mapValues(vs => Vectors.sparse(n, vs.toSeq))

现在您已经拥有创建 LabeledPoints 或应用其他处理所需的一切

回复于 2024-04-20T03:17:01+08:00

在火花中为LDA准备数据

1 回答

相关问题