Spark應用HanLP對中文語料進行文字挖掘--聚類詳解教程
軟體: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\t檔案內容
如: 41.txt 【 日 期 】199601....
這樣子的話,就可以通過.txt\t 來對每行文字進行分割,得到其檔名以及檔案內容,這裡每行其實就是一個檔案了。
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中加入這兩個依賴:
1. <!-- 中文分詞框架 -->
2.<dependency>
3.<groupId>com.hankcs</groupId>
4.<artifactId>hanlp</artifactId>
5.<version>${hanlp.version}</version>
6.</dependency>
7.<!-- Spark dependencies -->
8.<dependency>
9.<groupId>org.apache.spark</groupId>
10.<artifactId>spark-core_2.10</artifactId>
11.<version>${spark.version}</version>
12.</dependency>
13.<dependency>
14.<groupId>org.apache.spark</groupId>
15.<artifactId>spark-mllib_2.10</artifactId>
16.<version>${spark.version}</version>
17.</dependency>
其版本為:
<hanlp.version>portable-1.3.4</hanlp.version>、 <spark.version>1.6.0-cdh5.7.3</spark.version>。
3.2 檔案轉為UTF-8編碼及儲存到一個檔案
這部分內容可以直接參考:src/main/java/demo02_transform_encoding.TransformEncodingToOne 這裡的實現,因為是Java基本的操作,這裡就不加以分析了。
3.3 Scala呼叫HanLP進行中文分詞
Scala呼叫HanLP進行分詞和Java的是一樣的,同時,因為這裡有些詞語格式不正常,所以把這些特殊的詞語新增到自定義詞典中,其示例如下:
1.import com.hankcs.hanlp.dictionary.CustomDictionary
2.import com.hankcs.hanlp.dictionary.stopword.CoreStopWordDictionary
3.import com.hankcs.hanlp.tokenizer.StandardTokenizer
4.import scala.collection.JavaConversions._
5./**
6.* Scala 分詞測試
7.* Created by fansy on 2017/8/25.
8.*/
9.object SegmentDemo {
10.def main(args: Array[String]) {
11.val sentense = "41,【 日 期 】19960104 【 版 號 】1 【 標 題 】合巢蕪高速公路巢蕪段竣工 【 作 者 】彭建中 【 正 文 】 安徽合(肥)巢(湖)蕪(湖)高速公路巢蕪段日前竣工通車並投入營運。合巢蕪 高速公路是國家規劃的京福綜合運輸網的重要幹線路段,是交通部確定1995年建成 的全國10條重點公路之一。該條高速公路正線長88公里。(彭建中)"
12.CustomDictionary.add("日 期")
13.CustomDictionary.add("版 號")
14.CustomDictionary.add("標 題")
15.CustomDictionary.add("作 者")
16.CustomDictionary.add("正 文")
17.val list = StandardTokenizer.segment(sentense)
18.CoreStopWordDictionary.apply(list)
19.println(list.map(x => x.word.replaceAll(" ","")).mkString(","))
20.}
21.}
執行完成後,即可得到分詞的結果,如下:
考慮到使用方便,這裡把分詞封裝成一個函式:
1./**
2.* String 分詞
3.* @param sentense
4.* @return
5.*/
6.def transform(sentense:String):List[String] ={
7.val list = StandardTokenizer.segment(sentense)
8.CoreStopWordDictionary.apply(list)
9.list.map(x => x.word.replaceAll(" ","")).toList
10.}
11.}
輸入即是一箇中文的文字,輸出就是分詞的結果,同時去掉了一些常用的停用詞。
3.4 求TF-IDF
在Spark裡面求TF-IDF,可以直接呼叫Spark內建的演算法模組即可,同時在Spark的該演算法模組中還對求得的結果進行了維度變換(可以理解為特徵選擇或“降維”,當然這裡的降維可能是提升維度)。程式碼如下:
1.val docs = sc.textFile(input_data).map{x => val t = x.split(".txt\t");(t(0),transform(t(1)))}
2..toDF("fileName", "sentence_words")
3.
4.// 3. 求TF
5.println("calculating TF ...")
6.val hashingTF = new HashingTF()
7..setInputCol("sentence_words").setOutputCol("rawFeatures").setNumFeatures(numFeatures)
8.val featurizedData = hashingTF.transform(docs)
9.
10.// 4. 求IDF
11.println("calculating IDF ...")
12.val idf = new IDF().setInputCol("rawFeatures").setOutputCol("features")
13.val idfModel = idf.fit(featurizedData)
14.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模型
直接參考官網給定例子即可:
1.println("creating kmeans model ...")
2.val kmeans = new KMeans().setK(k).setSeed(1L)
3.val model = kmeans.fit(rescaledData)
4.// Evaluate clustering by computing Within Set Sum of Squared Errors.
5.println("calculating wssse ...")
6.val WSSSE = model.computeCost(rescaledData)
7.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中;
1.val data = sc.parallelize(t)
2.val file_index = data.map(_._1.charAt(0)).distinct.zipWithIndex().collect().toMap
3.println(file_index)
4.val partitionData = data.partitionBy(MyPartitioner(file_index))
這裡的file_index,是對不同類的文件進行編號,這個編號就對應每個partition,看MyPartitioner的實現:
1.case class MyPartitioner(file_index:Map[Char,Long]) extends Partitioner
2.override def getPartition(key: Any): Int = key match {
3.case _ => file_index.getOrElse(key.toString.charAt(0),0L).toInt
4.}
5..override def numPartitions: Int = file_index.size
6.}
2. 針對每個partition進行整合操作:
在整合每個partition之前,我們先看下我們自定義的MyPartitioner是否在正常工作,可以列印下結果:
1.val tt = partitionData.mapPartitionsWithIndex((index: Int, it: Iterator[(String,Int)]) => it.toList.map(x => (index,x)).toIterator)
2.tt.collect().foreach(println(_))
執行如下:
其中第一列代表每個partition的id,第二列是資料,發現其資料確實是按照預期進行處理的;接著可以針對每個partition進行資料整合:
1.// firstCharInFileName , firstCharInFileName - predictType
2.val combined = partitionData.map(x =>( (x._1.charAt(0), Integer.parseInt(x._1.charAt(0)+"") - x._2),1) )
3..mapPartitions{f => var aMap = Map[(Char,Int),Int]();
4.for(t <- f){
5.if (aMap.contains(t._1)){
6.aMap = aMap.updated(t._1,aMap.getOrElse(t._1,0)+1)
7.}else{
8.aMap = aMap + t
9.}
10.}
11.val aList = aMap.toList
12.val total= aList.map(_._2).sum
13.val total_right = aList.map(_._2).max
14.List((aList.head._1._1,total,total_right)).toIterator
15.// aMap.toIterator //列印各個partition的總結
16. }
在整合之前先執行一個map操作,把資料變成((fileNameFirstChar, fileNameFirstChar.toInt - predictId), 1),其中fileNameFirstChar代表檔案的第一個字元,其實也就是檔案的所屬實際類別,後面的fileNameFirstChar.toInt-predictId 其實就是判斷預測的結果是否對了,這個值的眾數就是預測對的;最後一個值程式碼前面的這個鍵值對出現的次數,其實就是統計屬於某個類別的實際檔案個數以及預測對的檔案個數,分別對應上面的total和total_right變數;輸出結果為:
(4,6,3)
(1,6,4)
(2,6,4)
發現其列印的結果是正確的,第一列代表檔名開頭,第二個代表屬於這個檔案的個數,第三列代表預測正確的個數
這裡需要注意的是,這裡因為文字的實際類別和檔名是一致的,所以才可以這樣處理,如果實際資料的話,那麼mapPartitions函式需要更改。
3. 針對資料結果進行統計:
最後只需要進行簡單的計算即可:
1.for(re <- result ){
2.println("文件"+re._1+"開頭的 文件總數:"+ re._2+",分類正確的有:"+re._3+",分類正確率是:"+(re._3*100.0/re._2)+"%")
3.}
4.val averageRate = result.map(_._3).sum *100.0 / result.map(_._2).sum
5.println("平均正確率為:"+averageRate+"%")
輸出結果為:
4. 實驗
設定不同的numFeature,比如使用200和2000,其對比結果為:
所以設定numFeatures值越大,其準確率也越高,不過計算也比較複雜。
5. 總結
1. HanLP的使用相對比較簡單,這裡只使用了分詞及停用詞,感謝開源;
2. Spark裡面的TF-IDF以及Word2Vector使用比較簡單,不過使用這個需要先分詞;
3. 這裡是在IDEA裡面執行的,如果使用Spark-submit的提交方式,那麼需要把hanpl的jar包加入,這個有待驗證
文章來源於fansy1990的部落格