引言

本文基于Spark (1.5.0) ml库提供的pipeline完整地实践一次文本分类。pipeline将串联单词分割(tokenize)、单词频数统计(TF)，特征向量计算(TF-IDF)，朴素贝叶斯（Naive Bayes）模型训练等。
本文将基于“20 NewsGroups” 数据集训练并测试Naive Bayes模型。这二十个新闻组数据集合是收集大约20,000新闻组文档，均匀的分布在20个不同的集合。我将使用'20news-bydate.tar.gz'文件，因为该数据集中已经将数据划分为两类：train和test，非常方便我们对模型进行训练和评价。

20news-bydate.tar.gz - 20 Newsgroups sorted by date; duplicates and some headers removed (18846 documents)

Naive Bayes算法介绍

NB算法属于有监督分类算法，对输入数据:

• 如果X属于某个分类y的概率的概率大于属于其它分类的概率，则判定该输入属于分类y;
• X属于某个分类y的概率正比于分类y自身出现的概率✖️该分类y条件下X各个维度出现的概率的乘积。

那么，模型训练的目标就很明朗了，我们需要基于给定的训练样本计算出:

• 各个分类的先验概率：
  
• 训练样本中，每个分类条件下，输入各个维度出现的似然概率：![][likeli_i]

模型用于分类新数据的计算：
![][predict]

[bayes]: http://latex.codecogs.com/svg.latex?\begin{align}P(y|(x_i,...,x_N))\=\frac{P(x_1,...,x_N|y)P(y)}{P(x_1,...,x_N)}\end{align}
[likeli]:http://latex.codecogs.com/svg.latex?P(x_1,...,x_N|y)=\prod_i^NP(x_i|y)
[likeli_i]:http://latex.codecogs.com/svg.latex?\theta(i,k)=P(x_i|y_k),i\in{1,...,N}
[predict]:http://latex.codecogs.com/svg.latex?y^{{new}=\text{arg}max_y{P(y)\cdot\prod_i}NP(x^{new}_i|y)}

spark mllib中算法流程

spark中对NaiveBayes算法的实现非常清晰明了，算法通过combineByKey计算每个分类下：
![][calc_pi]和![][calc_th_i_k]
[calc_pi]:http://latex.codecogs.com/svg.latex?p_k=\frac{\sum_{j=1}^{M\mathbb{I}(y}j=y_k)}{M}
[calc_th_i_k]: http://latex.codecogs.com/svg.latex?\theta(,k)=\frac{\sum_{j=1}^{M\mathbb{I}(y}j=y_k)\cdot{X^{j}+\alpha}{\sum_{j=1}}M{X^j}+\alpha\cdot{M}}

20 newsgroups实践

数据集分为train和test两组，分别用于训练和测试。每组数据都分为20类，每类数据存放在各自子文件下：

.
├── 20news-bydate-test
│   ├── alt.atheism
│   ├── comp.graphics
│   ├── comp.os.ms-windows.misc
│   ├── comp.sys.ibm.pc.hardware
│   ├── comp.sys.mac.hardware
│   ├── comp.windows.x
│   ├── misc.forsale
│   ├── rec.autos
│   ├── rec.motorcycles
│   ├── rec.sport.baseball
│   ├── rec.sport.hockey
│   ├── sci.crypt
│   ├── sci.electronics
│   ├── sci.med
│   ├── sci.space
│   ├── soc.religion.christian
│   ├── talk.politics.guns
│   ├── talk.politics.mideast
│   ├── talk.politics.misc
│   └── talk.religion.misc
└── 20news-bydate-train
    ├── alt.atheism
    ├── comp.graphics
    ├── comp.os.ms-windows.misc
    ├── comp.sys.ibm.pc.hardware
    ├── comp.sys.mac.hardware
    ├── comp.windows.x
    ├── misc.forsale
    ├── rec.autos
    ├── rec.motorcycles
    ├── rec.sport.baseball
    ├── rec.sport.hockey
    ├── sci.crypt
    ├── sci.electronics
    ├── sci.med
    ├── sci.space
    ├── soc.religion.christian
    ├── talk.politics.guns
    ├── talk.politics.mideast
    ├── talk.politics.misc
    └── talk.religion.misc

原始文档将经过如下流程训练得到NaiveBayes模型：

代码中的几点注解：

• 各类数据根据所在的子文件夹来分类，我们在写代码时需要利用子文件夹名称，这时可以通过调用sc.wholeTextFiles(...)函数得到RDD(String,String)类型的原始数据，_1表示文件的绝对路径，_2表示该文件的内容。我们进一步从_1中截取出子文件夹的名称f.split("/").takeRight(2).head.
• pipeline框架基于DataFrame，所有我们需要将RDD转为DataFrame：

import sqlContext.implicits._
labelNameAndData.toDF("id", "sentence").cache()```
- 所有的转换都使用ml提供的类，未做任何定制或改动，当前模型在测试集上的准确度为82%。

代码：
```scala
import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.classification.NaiveBayes
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
import org.apache.spark.ml.feature.{HashingTF, IDF, Tokenizer}
import org.apache.spark.{Logging, SparkConf, SparkContext}


object NBTest extends App with Logging {
  def createRawDf(s: String) = {
    //sc.setLogLevel("INFO")
    val fileNameData = sc.wholeTextFiles(s)

    val uniqueLabels = Array("alt.atheism", "comp.graphics", "comp.os.ms-windows.misc", "comp.sys.ibm.pc.hardware", "comp.sys.mac.hardware", "comp.windows.x", "misc.forsale", "rec.autos", "rec.motorcycles", "rec.sport.baseball", "rec.sport.hockey", "sci.crypt", "sci.electronics", "sci.med", "sci.space", "soc.religion.christian", "talk.politics.guns", "talk.politics.mideast", "talk.politics.misc", "talk.religion.misc")
    val uniqueLabelsBc = sc.broadcast(uniqueLabels)

    val labelNameAndData = fileNameData
      .map { case (f, data) => (f.split("/").takeRight(2).head, data) }
      .mapPartitions {
        itrs =>
          val labelIdMap = uniqueLabelsBc.value.zipWithIndex.toMap
          itrs.map {
            case (labelName, data) => (labelIdMap(labelName), data)
          }
      }

    import sqlContext.implicits._
    labelNameAndData.toDF("id", "sentence").cache()

  }

  def createTrainPpline() = {
    val tokenizer = new Tokenizer().setInputCol("sentence").setOutputCol("words")

    val hashingTF = new HashingTF().setInputCol("words").setOutputCol("rawFeatures")

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

    //val vecAssembler = new VectorAssembler().setInputCols(Array("features")).setOutputCol("id")

    val nb = new NaiveBayes().setFeaturesCol("features").setLabelCol("id")

    new Pipeline().setStages(Array(tokenizer, hashingTF, idf, nb))
  }

  val conf = new SparkConf().setMaster("local[2]").setAppName("nb")
    .set("spark.ui.enabled", "false")
  val sc = new SparkContext(conf)
  val sqlContext = new org.apache.spark.sql.SQLContext(sc)

  val training = createRawDf("file:////root/work/test/20news-bydate-train/*")

  val ppline = createTrainPpline()
  val nbModel = ppline.fit(training)

  val test = createRawDf("file:////root/work/test/20news-bydate-test/*")
  val testRes = nbModel.transform(test)

  val evaluator = new MulticlassClassificationEvaluator().setLabelCol("id")
  val accuracy = evaluator.evaluate(testRes)
  println("Test Error = " + (1.0 - accuracy))

}