電子發燒友App
軟件：IDEA2014、Maven、HanLP、JDK；
用到的知識：HanLP、Spark TF-IDF、Spark kmeans、Spark mapPartition;
用到的數據集：http://www.threedweb.cn/thread-1288-1-1.html（不需要下載，已經包含在工程里面）；
工程下載：https://github.com/fansy1990/hanlp-test 。
1. 問題描述
? ? 現在有一個中文文本數據集，這個數據集已經對其中的文本做了分類，如下：
其中每個文件夾中含有個數不等的文件，比如環境有200個，藝術有248個；同時，每個文件的內容基本上就是一些新聞報道或者中文描述，如下：
?
現在需要做的就是，把這些文檔進行聚類，看其和原始給定的類別的重合度有多少，這樣也可以反過來驗證我們聚類算法的正確度。
2. 解決思路：
? 2.1 文本預處理：
?1.? ? 由于文件的編碼是GBK的，讀取到Spark中全部是亂碼，所以先使用Java把代碼轉為UTF8編碼；??
?2. 由于文本存在多個文件中（大概2k多），使用Spark的wholeTextFile讀取速度太慢，所以考慮把這些文件全部合并為一個文件，這時又結合1.的轉變編碼，所以在轉變編碼的時候就直接把所有的數據存入同一個文件中；
? ? 其存儲的格式為： 每行：? ? 文件名.txt 文件內容
? ?如：? 41.txt 【 日? 期 】199601....
這樣子的話，就可以通過.txt 來對每行文本進行分割，得到其文件名以及文件內容，這里每行其實就是一個文件了。
2.2 分詞
? ?分詞直接采用HanLP的分詞來做，HanLP這里選擇兩種：Standard和NLP(還有一種就是HighSpeed，但是這個木有用戶自定義詞典，所以前期考慮先用兩種)，具體參考：https://github.com/hankcs/HanLP ;
2.3 詞轉換為詞向量
? 在Kmeans算法中，一個樣本需要使用數值類型，所以需要把文本轉為數值向量形式，這里在Spark中有兩種方式。其一，是使用TF-IDF；其二，使用Word2Vec。這里暫時使用了TF-IDF算法來進行，這個算法需要提供一個numFeatures，這個值越大其效果也越好，但是相應的計算時間也越長，后面也可以通過實驗驗證。
2.4 使用每個文檔的詞向量進行聚類建模
在進行聚類建模的時候，需要提供一個初始的聚類個數，這里面設置為10，因為我們的數據是有10個分組的。但是在實際的情況下，一般這個值是需要通過實驗來驗證得到的。
2.5 對聚類后的結果進行評估
這里面采用的思路是：
1. 得到聚類模型后，對原始數據進行分類，得到原始文件名和預測的分類id的二元組(fileName,predictId)；
2. 針對（fileName，predictId），得到（fileNameFirstChar ,fileNameFirstChar.toInt - predictId）的值，這里需要注意的是fileNameFirstChar其實就是代表這個文件的原始所屬類別了。
3. 這里有一個一般假設，就是使用kmeans模型預測得到的結果大多數是正確的，所以fileNameFirstChar.toInt-predictId得到的眾數其實就是分類的正確的個數了（這里可能比較難以理解，后面會有個小李子來說明這個問題）；
4. 得到每個實際類別的預測的正確率后就可以去平均預測率了。
5. 改變numFeatuers的值，看下是否numFeatures設置的比較大，其正確率也會比較大？
3. 具體步驟：
3.1 開發環境--Maven
首先第一步，當然是開發環境了，因為用到了Spark和HanLP，所以需要在pom.xml中加入這兩個依賴：
? ? ? ?
? ? ? ? ? ?
? ? ? ? ? ?
? ? ? ? ? ?
? ? ? ?
?
? ? ? ?
? ? ? ?
? ? ? ? ? ?
? ? ? ? ? ?
? ? ? ? ? ?
? ? ? ?
? ? ? ?
? ? ? ? ? ?
? ? ? ? ? ?
? ? ? ? ? ?
? ? ? ?
其版本為：
3.2 文件轉為UTF-8編碼及存儲到一個文件
這部分內容可以直接參考：src/main/java/demo02_transform_encoding.TransformEncodingToOne 這里的實現，因為是Java基本的操作，這里就不加以分析了。
3.3 Scala調用HanLP進行中文分詞
Scala調用HanLP進行分詞和Java的是一樣的，同時，因為這里有些詞語格式不正常，所以把這些特殊的詞語添加到自定義詞典中，其示例如下：
import com.hankcs.hanlp.dictionary.CustomDictionary
import com.hankcs.hanlp.dictionary.stopword.CoreStopWordDictionary
import com.hankcs.hanlp.tokenizer.StandardTokenizer
import scala.collection.JavaConversions._
/**
?* Scala 分詞測試
?* Created by fansy on 2017/8/25.
?*/
object SegmentDemo {
? def main(args: Array[String]) {
? ? val sentense = "41,【 日? 期 】19960104 【 版? 號 】1 【 標? 題 】合巢蕪高速公路巢蕪段竣工 【 作? 者 】彭建中 【 正? 文 】? ? ?安徽合（肥）巢（湖）蕪（湖）高速公路巢蕪段日前竣工通車并投入營運。合巢蕪 高速公路是國家規劃的京福綜合運輸網的重要干線路段，是交通部確定１９９５年建成 的全國１０條重點公路之一。該條高速公路正線長８８公里。（彭建中）"
? ? CustomDictionary.add("日? 期")
? ? CustomDictionary.add("版? 號")
? ? CustomDictionary.add("標? 題")
? ? CustomDictionary.add("作? 者")
? ? CustomDictionary.add("正? 文")
? ? val list = StandardTokenizer.segment(sentense)
? ? CoreStopWordDictionary.apply(list)
? ? println(list.map(x => x.word.replaceAll(" ","")).mkString(","))
? }
}
運行完成后，即可得到分詞的結果，如下：
考慮到使用方便，這里把分詞封裝成一個函數：
?/**
? ?* String 分詞
? ?* @param sentense
? ?* @return
? ?*/
? def transform(sentense:String):List[String] ={
? ? val list = StandardTokenizer.segment(sentense)
? ? CoreStopWordDictionary.apply(list)
? ? list.map(x => x.word.replaceAll(" ","")).toList
? }
?}
輸入即是一個中文的文本，輸出就是分詞的結果，同時去掉了一些常用的停用詞。
3.4 求TF-IDF
在Spark里面求TF-IDF，可以直接調用Spark內置的算法模塊即可，同時在Spark的該算法模塊中還對求得的結果進行了維度變換（可以理解為特征選擇或“降維”，當然這里的降維可能是提升維度）。代碼如下：
val docs = sc.textFile(input_data).map{x => val t = x.split(".txt ");(t(0),transform(t(1)))}
? ? ? ?.toDF("fileName", "sentence_words")
?
? ? ?// 3. 求TF
? ? ?println("calculating TF ...")
? ? ?val hashingTF = new HashingTF()
? ? ? ?.setInputCol("sentence_words").setOutputCol("rawFeatures").setNumFeatures(numFeatures)
? ? ?val featurizedData = hashingTF.transform(docs)
?
? ? ?// 4. 求IDF
? ? ?println("calculating IDF ...")
? ? ?val idf = new IDF().setInputCol("rawFeatures").setOutputCol("features")
? ? ?val idfModel = idf.fit(featurizedData)
? ? ?val rescaledData = idfModel.transform(featurizedData).cache()
變量docs是一個DataFrame[fileName, sentence_words] ,經過HashingTF后，變成了變量 featurizedData ,同樣是一個DataFrame[fileName,sentence_words, rawFeatures]。這里通過setInputCol以及SetOutputCol可以設置輸入以及輸出列名（列名是針對DataFrame來說的，不知道的可以看下DataFrame的API）。
接著，經過IDF模型，得到變量 rescaledData ，其DataFrame[fileName,sentence_words, rawFeatures, features] 。
執行結果為：
3.5 建立KMeans模型
直接參考官網給定例子即可：
println("creating kmeans model ...")
? ? ?val kmeans = new KMeans().setK(k).setSeed(1L)
? ? ?val model = kmeans.fit(rescaledData)
? ? ?// Evaluate clustering by computing Within Set Sum of Squared Errors.
? ? ?println("calculating wssse ...")
? ? ?val WSSSE = model.computeCost(rescaledData)
? ? ?println(s"Within Set Sum of Squared Errors = $WSSSE")
這里有計算cost值的，但是這個值評估不是很準確，比如我numFeature設置為2000的話，那么這個值就很大，但是其實其正確率會比較大的。
3.6 模型評估
這里的模型評估直接使用一個小李子來說明：比如，現在有這樣的數據：
其中，1開頭，2開頭和4開頭的屬于同一類文檔，后面的0,3,2,1等，代表這個文檔被模型分類的結果，那么可以很容易的看出針對1開頭的文檔，
其分類正確的有4個，其中("123.txt",3)以及（“126.txt”,1）是分類錯誤的結果，這是因為，在這個類別中預測的結果中0是最多的，所以0是和1開頭的文檔對應起來的，這也就是前面的假設。
1. 把同一類文檔分到同一個partition中；
val data = sc.parallelize(t)
? ? ?val file_index = data.map(_._1.charAt(0)).distinct.zipWithIndex().collect().toMap
? ? ?println(file_index)
? ? ?val partitionData = data.partitionBy(MyPartitioner(file_index))
這里的file_index，是對不同類的文檔進行編號，這個編號就對應每個partition，看MyPartitioner的實現：
case class MyPartitioner(file_index:Map[Char,Long]) extends Partitioner{
? override def getPartition(key: Any): Int = key match {
? ? case _ => file_index.getOrElse(key.toString.charAt(0),0L).toInt
? }
? override def numPartitions: Int = file_index.size
}
2. 針對每個partition進行整合操作：
在整合每個partition之前，我們先看下我們自定義的MyPartitioner是否在正常工作，可以打印下結果：
val tt = partitionData.mapPartitionsWithIndex((index: Int, it: Iterator[(String,Int)]) => it.toList.map(x => (index,x)).toIterator)
? ? ?tt.collect().foreach(println(_))
運行如下：
其中第一列代表每個partition的id，第二列是數據，發現其數據確實是按照預期進行處理的；接著可以針對每個partition進行數據整合：
// firstCharInFileName , firstCharInFileName - predictType
? ? ?val combined = partitionData.map(x =>( (x._1.charAt(0), Integer.parseInt(x._1.charAt(0)+"") - x._2),1) )
? ? ?.mapPartitions{f => var aMap = Map[(Char,Int),Int]();
? ? ? ?for(t <- f){
? ? ? ? ?if (aMap.contains(t._1)){
? ? ? ? ? ?aMap = aMap.updated(t._1,aMap.getOrElse(t._1,0)+1)
? ? ? ? ?}else{
? ? ? ? ? ?aMap = aMap + t
? ? ? ? ?}
? ? ? ?}
? ? ? ?val aList = aMap.toList
? ? ? ?val total= aList.map(_._2).sum
? ? ? ?val total_right = aList.map(_._2).max
? ? ? ?List((aList.head._1._1,total,total_right)).toIterator
? ? ? ?//? ? ? ?aMap.toIterator //打印各個partition的總結
? ? ?}
在整合之前先執行一個map操作，把數據變成（(fileNameFirstChar, fileNameFirstChar.toInt - predictId), 1），其中fileNameFirstChar代表文件的第一個字符，其實也就是文件的所屬實際類別，后面的fileNameFirstChar.toInt-predictId 其實就是判斷預測的結果是否對了，這個值的眾數就是預測對的；最后一個值代碼前面的這個鍵值對出現的次數，其實就是統計屬于某個類別的實際文件個數以及預測對的文件個數，分別對應上面的total和total_right變量；輸出結果為：
(4,6,3)
(1,6,4)
(2,6,4)
發現其打印的結果是正確的，第一列代表文件名開頭，第二個代表屬于這個文件的個數，第三列代表預測正確的個數
這里需要注意的是，這里因為文本的實際類別和文件名是一致的，所以才可以這樣處理，如果實際數據的話，那么mapPartitions函數需要更改。
3. 針對數據結果進行統計：
最后只需要進行簡單的計算即可：
for(re <- result ){
? ? ? ? println("文檔"+re._1+"開頭的 文檔總數："+ re._2+",分類正確的有："+re._3+",分類正確率是："+(re._3*100.0/re._2)+"%")
? ? ? }
? ? ?val averageRate = result.map(_._3).sum *100.0 / result.map(_._2).sum
? ? ?println("平均正確率為："+averageRate+"%")
輸出結果為：
4. 實驗
? 設置不同的numFeature，比如使用200和2000，其對比結果為：
所以設置numFeatures值越大，其準確率也越高，不過計算也比較復雜。
5. 總結
1. HanLP的使用相對比較簡單，這里只使用了分詞及停用詞，感謝開源；
2. Spark里面的TF-IDF以及Word2Vector使用比較簡單，不過使用這個需要先分詞；
3. 這里是在IDEA里面運行的，如果使用Spark-submit的提交方式，那么需要把hanpl的jar包加入，這個有待驗證；
文章來源于fansy1990的博客
工商網監
評論