中文分词

对于中文文本分类而言，需要先对文章进行分词，我使用的是IKAnalyzer中文分析工具，其中自己可以配置扩展词库来使分词结果更合理，我从搜狗、百度输入法下载了细胞词库，将其作为扩展词库。这里不再介绍分词。

中文词语特征值转换（TF-IDF）

分好词后，每一个词都作为一个特征，但需要将中文词语转换成Double型来表示，通常使用该词语的TF-IDF值作为特征值，Spark提供了全面的特征抽取及转换的API，非常方便，详见http://spark.apache.org/docs/latest/ml-features.html,这里介绍下TF-IDF的API:

比如，训练语料/tmp/lxw1234/1.txt：

0,苹果 官网 苹果 宣布
1,苹果 梨 香蕉

逗号分隔的第一列为分类编号，0为科技，1为水果。

case class RawDataRecord(category: String, text: String)val conf = new SparkConf().setMaster("yarn-client")
val sc = new SparkContext(conf)
val sqlContext = new org.apache.spark.sql.SQLContext(sc)
import sqlContext.implicits._//将原始数据映射到DataFrame中，字段category为分类编号，字段text为分好的词，以空格分隔
var srcDF = sc.textFile("/tmp/lxw1234/1.txt").map { x => var data = x.split(",")RawDataRecord(data(0),data(1))
}.toDF()srcDF.select("category", "text").take(2).foreach(println)
[0,苹果 官网 苹果 宣布]
[1,苹果 梨 香蕉]
//将分好的词转换为数组
var tokenizer = new Tokenizer().setInputCol("text").setOutputCol("words")
var wordsData = tokenizer.transform(srcDF)wordsData.select($"category",$"text",$"words").take(2).foreach(println)
[0,苹果 官网 苹果 宣布,WrappedArray(苹果, 官网, 苹果, 宣布)]
[1,苹果 梨 香蕉,WrappedArray(苹果, 梨, 香蕉)]//将每个词转换成Int型，并计算其在文档中的词频（TF）
var hashingTF =
new HashingTF().setInputCol("words").setOutputCol("rawFeatures").setNumFeatures(100)
var featurizedData = hashingTF.transform(wordsData)

这里将中文词语转换成INT型的Hashing算法，类似于Bloomfilter，上面的setNumFeatures(100)表示将Hash分桶的数量设置为100个，这个值默认为2的20次方，即1048576，可以根据你的词语数量来调整，一般来说，这个值越大，不同的词被计算为一个Hash值的概率就越小，数据也更准确，但需要消耗更大的内存，和Bloomfilter是一个道理。

featurizedData.select($"category", $"words", $"rawFeatures").take(2).foreach(println)
[0,WrappedArray(苹果, 官网, 苹果, 宣布),(100,[23,81,96],[2.0,1.0,1.0])]
[1,WrappedArray(苹果, 梨, 香蕉),(100,[23,72,92],[1.0,1.0,1.0])]

结果中，“苹果”用23来表示，第一个文档中，词频为2，第二个文档中词频为1.

//计算TF-IDF值
var idf = new IDF().setInputCol("rawFeatures").setOutputCol("features")
var idfModel = idf.fit(featurizedData)
var rescaledData = idfModel.transform(featurizedData)
rescaledData.select($"category", $"words", $"features").take(2).foreach(println)[0,WrappedArray(苹果, 官网, 苹果, 宣布),(100,[23,81,96],[0.0,0.4054651081081644,0.4054651081081644])]
[1,WrappedArray(苹果, 梨, 香蕉),(100,[23,72,92],[0.0,0.4054651081081644,0.4054651081081644])]//因为一共只有两个文档，且都出现了“苹果”，因此该词的TF-IDF值为0.

最后一步，将上面的数据转换成Bayes算法需要的格式，如：

https://github.com/apache/spark/blob/branch-1.5/data/mllib/sample_naive_bayes_data.txtvar trainDataRdd = rescaledData.select($"category",$"features").map {case Row(label: String, features: Vector) =>LabeledPoint(label.toDouble, Vectors.dense(features.toArray))
}

每一个LabeledPoint中，特征数组的长度为100（setNumFeatures(100)），”官网”和”宣布”对应的特征索引号分别为81和96，因此，在特征数组中，第81位和第96位分别为它们的TF-IDF值。

到此，中文词语特征表示的工作已经完成，trainDataRdd已经可以作为Bayes算法的输入了。

分类模型训练

训练模型，语料非常重要，我这里使用的是搜狗提供的分类语料库，很早之前的了，这里只作为学习测试使用。

下载地址在：http://www.sogou.com/labs/dl/c.html，语料库一共有10个分类：

C000007 汽车
C000008 财经
C000010  IT
C000013 健康
C000014 体育
C000016 旅游
C000020 教育
C000022 招聘
C000023 文化
C000024 军事

每个分类下有几千个文档，这里将这些语料进行分词，然后每一个分类生成一个文件，在该文件中，每一行数据表示一个文档的分词结果，重新用0-9作为这10个分类的编号：

0 汽车
1 财经
2 IT
3 健康
4 体育
5 旅游
6 教育
7 招聘
8 文化
9 军事

比如，汽车分类下的文件内容为：

数据准备好了，接下来进行模型训练及分类预测,代码：

package com.lxw1234.textclassificationimport scala.reflect.runtime.universeimport org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.ml.feature.HashingTF
import org.apache.spark.ml.feature.IDF
import org.apache.spark.ml.feature.Tokenizer
import org.apache.spark.mllib.classification.NaiveBayes
import org.apache.spark.mllib.linalg.Vector
import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.mllib.regression.LabeledPoint
import org.apache.spark.sql.Rowobject TestNativeBayes {case class RawDataRecord(category: String, text: String)def main(args : Array[String]) {val conf = new SparkConf().setMaster("yarn-client")val sc = new SparkContext(conf)val sqlContext = new org.apache.spark.sql.SQLContext(sc)import sqlContext.implicits._var srcRDD = sc.textFile("/tmp/lxw1234/sougou/").map { x => var data = x.split(",")RawDataRecord(data(0),data(1))}//70%作为训练数据，30%作为测试数据val splits = srcRDD.randomSplit(Array(0.7, 0.3))var trainingDF = splits(0).toDF()var testDF = splits(1).toDF()//将词语转换成数组var tokenizer = new Tokenizer().setInputCol("text").setOutputCol("words")var wordsData = tokenizer.transform(trainingDF)println("output1：")wordsData.select($"category",$"text",$"words").take(1)//计算每个词在文档中的词频var hashingTF = new HashingTF().setNumFeatures(500000).setInputCol("words").setOutputCol("rawFeatures")var featurizedData = hashingTF.transform(wordsData)println("output2：")featurizedData.select($"category", $"words", $"rawFeatures").take(1)//计算每个词的TF-IDFvar idf = new IDF().setInputCol("rawFeatures").setOutputCol("features")var idfModel = idf.fit(featurizedData)var rescaledData = idfModel.transform(featurizedData)println("output3：")rescaledData.select($"category", $"features").take(1)//转换成Bayes的输入格式var trainDataRdd = rescaledData.select($"category",$"features").map {case Row(label: String, features: Vector) =>LabeledPoint(label.toDouble, Vectors.dense(features.toArray))}println("output4：")trainDataRdd.take(1)//训练模型val model = NaiveBayes.train(trainDataRdd, lambda = 1.0, modelType = "multinomial")   //测试数据集，做同样的特征表示及格式转换var testwordsData = tokenizer.transform(testDF)var testfeaturizedData = hashingTF.transform(testwordsData)var testrescaledData = idfModel.transform(testfeaturizedData)var testDataRdd = testrescaledData.select($"category",$"features").map {case Row(label: String, features: Vector) =>LabeledPoint(label.toDouble, Vectors.dense(features.toArray))}//对测试数据集使用训练模型进行分类预测val testpredictionAndLabel = testDataRdd.map(p => (model.predict(p.features), p.label))//统计分类准确率var testaccuracy = 1.0 * testpredictionAndLabel.filter(x => x._1 == x._2).count() / testDataRdd.count()println("output5：")println(testaccuracy)}
}

执行后，主要输出如下：

output1：（将词语转换成数组）

output2：（计算每个词在文档中的词频）

output3：（计算每个词的TF-IDF）

output4：（Bayes算法的输入数据格式）

output5：（测试数据集分类准确率）

准确率90%，还可以。接下来需要收集分类更细，时间更新的数据来训练和测试了。