如何准备mllib中的训练数据

Jus*_*ony 5 apache-spark apache-spark-ml apache-spark-mllib

TL; DR; 我如何使用mllib培训我的维基数据(文本和类别)来预测推文?

我有麻烦搞清楚如何我的标记化维基数据转换,以便它可以通过任何的培训NaiveBayesLogisticRegression.我的目标是使用经过训练的模型与推文进行比较*.我已经使用管道与LR,并试图HashingTFIDFNaiveBayes,但我不断收到错误的预测.这是我尝试过的:

*请注意,我想在我的标签中使用维基数据中的许多类别...我只看到二元分类(它是一个或另一个类别)....是否可以做我想要的?

管道w LR

import org.apache.spark.rdd.RDD
import org.apache.spark.SparkContext
import org.apache.spark.ml.feature.HashingTF
import org.apache.spark.mllib.linalg.Vector
import org.apache.spark.ml.feature.RegexTokenizer

case class WikiData(category: String, text: String)
case class LabeledData(category: String, text: String, label: Double)

val wikiData = sc.parallelize(List(WikiData("Spark", "this is about spark"), WikiData("Hadoop","then there is hadoop")))

val categoryMap = wikiData.map(x=>x.category).distinct.zipWithIndex.mapValues(x=>x.toDouble/1000).collectAsMap

val labeledData = wikiData.map(x=>LabeledData(x.category, x.text, categoryMap.get(x.category).getOrElse(0.0))).toDF

val tokenizer = new RegexTokenizer()
  .setInputCol("text")
  .setOutputCol("words")
  .setPattern("/W+")
val hashingTF = new HashingTF()
  .setNumFeatures(1000)
  .setInputCol(tokenizer.getOutputCol)
  .setOutputCol("features")
val lr = new LogisticRegression()
  .setMaxIter(10)
  .setRegParam(0.01)
val pipeline = new Pipeline()
  .setStages(Array(tokenizer, hashingTF, lr))

val model = pipeline.fit(labeledData)

model.transform(labeledData).show
Run Code Online (Sandbox Code Playgroud)

朴素贝叶斯

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

import org.apache.spark.mllib.feature.IDF

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

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

//to create tfidfLabeled (below) I ran a map set the labels...but again it seems to have to be 1.0 or 0.0?

NaiveBayes.train(tfidfLabeled)
  .predict(hashingTF.transform(tweet))
  .collect
Run Code Online (Sandbox Code Playgroud)

zer*_*323 3

MLLogisticRegression尚不支持多项分类,但 MLLibNaiveBayesLogisticRegressionWithLBFGS. 在第一种情况下,它应该默认工作:

import org.apache.spark.mllib.classification.NaiveBayes

val nbModel = new NaiveBayes()
  .setModelType("multinomial") // This is default value
  .run(train)
Run Code Online (Sandbox Code Playgroud)

但对于逻辑回归,您应该提供许多类:

import org.apache.spark.mllib.classification.LogisticRegressionWithLBFGS

val model = new LogisticRegressionWithLBFGS()
  .setNumClasses(n) // Set number of classes
  .run(trainingData)
Run Code Online (Sandbox Code Playgroud)

关于预处理步骤,这是一个相当广泛的主题,如果不访问您的数据,很难给您提供有意义的建议,因此您在下面找到的所有内容都只是一个疯狂的猜测:

  • 据我了解,您使用 wiki 数据进行训练,使用推文进行测试。如果这是真的,那么一般来说这是一个坏主意。您可以预期这两个集合使用的词汇、语法和拼写有显着不同
  • 简单的正则表达式标记生成器可以在标准化文本上表现得很好,但根据我的经验,它在推文等非正式文本上效果不佳
  • HashingTF可能是获取基线模型的好方法,但它是极其简化的方法,特别是如果您不应用任何过滤步骤。如果您决定使用它,您至少应该增加功能数量或使用默认值(2^20)

编辑(使用 IDF 为朴素贝叶斯准备数据)

  • 使用机器学习管道:
import org.apache.spark.mllib.regression.LabeledPoint
import org.apache.spark.mllib.linalg.Vector
import org.apache.spark.ml.feature.IDF
import org.apache.spark.sql.Row

val tokenizer = ???

val hashingTF = new HashingTF()
  .setNumFeatures(1000)
  .setInputCol(tokenizer.getOutputCol)
  .setOutputCol("rawFeatures")

val idf = new IDF()
  .setInputCol(hashingTF.getOutputCol)
  .setOutputCol("features")

val pipeline = new Pipeline().setStages(Array(tokenizer, hashingTF, idf))
val model = pipeline.fit(labeledData)

model
 .transform(labeledData)
 .select($"label", $"features")
 .map{case Row(label: Double, features: Vector) => LabeledPoint(label, features)}
Run Code Online (Sandbox Code Playgroud)
  • 使用 MLlib 转换器:
import org.apache.spark.mllib.feature.HashingTF
import org.apache.spark.mllib.linalg.Vector
import org.apache.spark.mllib.feature.{IDF, IDFModel}

val labeledData = wikiData.map(x => 
  LabeledData(x.category, x.text, categoryMap.get(x.category).getOrElse(0.0)))

val p = "\\W+".r
val raw = labeledData.map{
    case LabeledData(_, text, label) => (label, p.split(text))}

val hashingTF: org.apache.spark.mllib.feature.HashingTF = new HashingTF(1000)
val tf = raw.map{case (label, text) => (label, hashingTF.transform(text))}

val idf: org.apache.spark.mllib.feature.IDFModel = new IDF().fit(tf.map(_._2))
tf.map{
  case (label, rawFeatures) => LabeledPoint(label, idf.transform(rawFeatures))}
Run Code Online (Sandbox Code Playgroud)

注意:由于转换器需要 JVM 访问,因此 MLlib 版本无法在 PySpark 中工作。如果您更喜欢 Python,则必须拆分 data conversion 和 zip

编辑(为机器学习算法准备数据):

虽然下面的代码乍一看似乎有效

val categoryMap = wikiData
  .map(x=>x.category)
  .distinct
  .zipWithIndex
  .mapValues(x=>x.toDouble/1000)
  .collectAsMap

val labeledData = wikiData.map(x=>LabeledData(
    x.category, x.text, categoryMap.get(x.category).getOrElse(0.0))).toDF
Run Code Online (Sandbox Code Playgroud)

它不会为ML算法生成有效的标签。

首先,ML期望标签位于 (0.0, 1.0, ..., n.0) 中,其中 n 是类别数。如果您的示例管道中其中一个类的标签为 0.001,您将收到如下错误:

错误 LogisticRegression:分类标签应位于 {0 到 0 发现 1 个无效标签。

显而易见的解决方案是在生成映射时避免除法

.mapValues(x=>x.toDouble)
Run Code Online (Sandbox Code Playgroud)

虽然它适用于LogisticRegression其他ML算法,但仍然会失败。例如RandomForestClassifier你会得到

RandomForestClassifier 的输入带有无效的标签列标签,但没有指定类的数量。请参阅字符串索引器。

有趣的是RandomForestClassifier,ML 版本与它的对应版本不同MLlib,它没有提供设置多个类的方法。事实证明,它需要在列上设置特殊属性DataFrame。最简单的方法是使用StringIndexer错误消息中提到的:

.mapValues(x=>x.toDouble)
Run Code Online (Sandbox Code Playgroud)

  • 您可以使用 ML 或 MLlib。我已经编辑了答案。 (2认同)
  • 万岁!这绝对是我在 Spark 中最薄弱的领域,也是我将在未来几年大力加强的领域:)。然而,对于我现在需要的东西来说,这太棒了!我会尽量记住明天添加赏金,为你的努力提供额外的积分。 (2认同)