Spark MLlib - 从 RDD[Vector] 特征和 RDD[Vector] 标签创建 LabeledPoint

Posted 2023-04-15

【中文标题】Spark MLlib - 从 RDD[Vector] 特征和 RDD[Vector] 标签创建 LabeledPoint

【英文标题】：Spark MLib - Create LabeledPoint from RDD[Vector] features and RDD[Vector] label

【发布时间】：2016-03-11 21:36:00

【问题描述】：

我正在使用两个代表文档和标签的文本文件构建一个训练集。

Documents.txt

hello world
hello mars

Labels.txt

0
1

我已阅读这些文件并将我的文档数据转换为tf-idf 加权term-document matrix，表示为RDD[Vector]。我还阅读并为我的标签创建了RDD[Vector]：

val docs: RDD[Seq[String]] = sc.textFile("Documents.txt").map(_.split(" ").toSeq)
val labs: RDD[Vector] = sc.textFile("Labels.txt")
  .map(s => Vectors.dense(s.split(',').map(_.toDouble)))

val hashingTF = new HashingTF()
val tf: RDD[Vector] = hashingTF.transform(docs)
tf.cache()

val idf = new IDF(minDocFreq = 3).fit(tf)
val tfidf: RDD[Vector] = idf.transform(tf)

我想使用tfidf 和labs 来创建RDD[LabeledPoint]，但我不确定如何应用具有两个不同RDDs 的映射。这是否可能/有效，还是我需要重新考虑我的方法？

【问题讨论】：

你应该joinRDDs。

@AlbertoBonsanto 我正在考虑这种方法，但是如果RDD 没有keys 到join by 我该怎么做？

【参考方案1】：

处理此问题的一种方法是根据索引发送join：

import org.apache.spark.RangePartitioner

// Add indices
val idfIndexed = idf.zipWithIndex.map(_.swap)
val labelsIndexed = labels.zipWithIndex.map(_.swap)

// Create range partitioner on larger RDD
val partitioner = new RangePartitioner(idfIndexed.partitions.size, idfIndexed)

// Join with custom partitioner
labelsIndexed.join(idfIndexed, partitioner).values