Spark多类分类示例

Question

你们知道我在哪里可以找到Spark中多类分类的例子。我花了很多时间在书本和网络上搜索，到目前为止，我只知道根据文档的最新版本，这是可能的。

Answer 1

ML

（建议在星火2.0+）

我们将使用相同的数据如下MLlib。有两个基本选项。如果模型只支持二元分类（logistic回归），并延伸

val trainRawDf = trainRaw.toDF 

import org.apache.spark.ml.feature.{Tokenizer, CountVectorizer, StringIndexer} 
import org.apache.spark.ml.Pipeline 

import org.apache.spark.ml.classification.RandomForestClassifier 

val transformers = Array(
    new StringIndexer().setInputCol("group").setOutputCol("label"), 
    new Tokenizer().setInputCol("text").setOutputCol("tokens"), 
    new CountVectorizer().setInputCol("tokens").setOutputCol("features") 
) 


val rf = new RandomForestClassifier() 
    .setLabelCol("label") 
    .setFeaturesCol("features") 

val model = new Pipeline().setStages(transformers :+ rf).fit(trainRawDf) 

model.transform(trainRawDf) 

o.a.s.ml.classification.Classifier你可以使用一个-VS：如果Estimator支持multilclass分类外的开箱（例如随机森林），你可以直接使用它-rest策略：

import org.apache.spark.ml.classification.OneVsRest 
import org.apache.spark.ml.classification.LogisticRegression 

val lr = new LogisticRegression() 
    .setLabelCol("label") 
    .setFeaturesCol("features") 

val ovr = new OneVsRest().setClassifier(lr) 

val ovrModel = new Pipeline().setStages(transformers :+ ovr).fit(trainRawDf) 

MLLib

按照official documentation在这个时刻（MLlib 1.6.0）以下方法suppo RT多分类：

• 回归，
• 决策树，
• 随机森林，
• 朴素贝叶斯

至少一些例子使用多分类：

• Naive Bayes example - 3类
• Logistic regression - 10类分类器虽然只有2中的示例性数据

一般框架，忽略方法具体参数，几乎是相同的MLlib的所有其他方法。你必须预先处理您的输入来创建与代表label和features列，列数据帧：

root 
|-- label: double (nullable = true) 
|-- features: vector (nullable = true) 

或RDD[LabeledPoint]。

Spark提供了广泛的有用的工具，旨在促进此过程，包括Feature Extractors和Feature Transformers和pipelines。

你会发现一个相当天真的使用下面的随机森林的例子。

首先让进口所需的软件包，并创建虚拟数据：

import sqlContext.implicits._ 
import org.apache.spark.ml.feature.{HashingTF, Tokenizer} 
import org.apache.spark.mllib.regression.LabeledPoint 
import org.apache.spark.ml.feature.StringIndexer 
import org.apache.spark.mllib.tree.RandomForest 
import org.apache.spark.mllib.tree.model.RandomForestModel 
import org.apache.spark.mllib.linalg.{Vectors, Vector} 
import org.apache.spark.mllib.evaluation.MulticlassMetrics 
import org.apache.spark.sql.Row 
import org.apache.spark.rdd.RDD 

case class LabeledRecord(group: String, text: String) 

val trainRaw = sc.parallelize(
    LabeledRecord("foo", "foo v a y b foo") :: 
    LabeledRecord("bar", "x bar y bar v") :: 
    LabeledRecord("bar", "x a y bar z") :: 
    LabeledRecord("foobar", "foo v b bar z") :: 
    LabeledRecord("foo", "foo x") :: 
    LabeledRecord("foobar", "z y x foo a b bar v") :: 
    Nil 
) 

现在让我们来定义所需的变压器和加工列Dataset：

// Tokenizer to process text fields 
val tokenizer = new Tokenizer() 
    .setInputCol("text") 
    .setOutputCol("words") 

// HashingTF to convert tokens to the feature vector 
val hashingTF = new HashingTF() 
    .setInputCol("words") 
    .setOutputCol("features") 
    .setNumFeatures(10) 

// Indexer to convert String labels to Double 
val indexer = new StringIndexer() 
    .setInputCol("group") 
    .setOutputCol("label") 
    .fit(trainRaw.toDF) 


def transfom(rdd: RDD[LabeledRecord]) = { 
    val tokenized = tokenizer.transform(rdd.toDF) 
    val hashed = hashingTF.transform(tokenized) 
    val indexed = indexer.transform(hashed) 
    indexed 
     .select($"label", $"features") 
     .map{case Row(label: Double, features: Vector) => 
      LabeledPoint(label, features)} 
} 

val train: RDD[LabeledPoint] = transfom(trainRaw) 

请注意，indexer被“装”在火车上的数据。它只是意味着将用作标签的分类值转换为doubles。要在新数据上使用分类器，必须先使用此indexer对其进行转换。

接下来，我们可以训练RF模型：

val numClasses = 3 
val categoricalFeaturesInfo = Map[Int, Int]() 
val numTrees = 10 
val featureSubsetStrategy = "auto" 
val impurity = "gini" 
val maxDepth = 4 
val maxBins = 16 

val model = RandomForest.trainClassifier(
    train, numClasses, categoricalFeaturesInfo, 
    numTrees, featureSubsetStrategy, impurity, 
    maxDepth, maxBins 
) 

最后测试：

val testRaw = sc.parallelize(
    LabeledRecord("foo", "foo foo z z z") :: 
    LabeledRecord("bar", "z bar y y v") :: 
    LabeledRecord("bar", "a a bar a z") :: 
    LabeledRecord("foobar", "foo v b bar z") :: 
    LabeledRecord("foobar", "a foo a bar") :: 
    Nil 
) 

val test: RDD[LabeledPoint] = transfom(testRaw) 

val predsAndLabs = test.map(lp => (model.predict(lp.features), lp.label)) 
val metrics = new MulticlassMetrics(predsAndLabs) 

metrics.precision 
metrics.recall 

Answer 2

您正在使用的Spark 1.6，而不是2.1火花？ 我认为问题在于spark 2.1中的transform方法返回一个数据集，该数据集可以隐式转换为类型化的RDD，在此之前，它返回一个数据帧或行。

尝试作为RDD [LabeledPoint]指定转换函数的返回类型的诊断，并查看是否得到相同的错误。