SparkMllib主題模型案例講解

一  本文涉及到的算法

1， LDA主題模型

符號定義

• 文檔集合D，m篇，topic集合T，k個主題

• D中每一個文檔d看做一個單詞序列< w1,w2,...,wn >，wi表示第i個單詞，設d有n個單詞。（LDA裏面稱之爲word bag，實際上每一個單詞的出現位置對LDA算法無影響）

• D中涉及的全部不一樣單詞組成一個大集合VOCABULARY（簡稱VOC）

LDA符合的分佈

• 每篇文章d(長度爲)都有各自的主題分佈，主題分佈式多項分佈，該多項分佈的參數服從Dirichlet分佈，該Dirichlet分佈的參數爲α。

• 每一個主題都有各自的詞分佈，詞分佈爲多項分佈，該多項分佈的參數服從Dirichlet分佈，該Dirichlet分佈的參數爲β；

• 對於謀篇文章中的第n個詞，首先從該文章的主題分佈中採樣一個主題，而後在這個主題對應的詞分佈中採樣一個詞。不斷重複這個隨機生成過程，直到m篇文章所有完成過程。

結果是但願訓練出兩個結果向量(k個topic，VOC中共包含m個詞)

LDA以文檔集合D做爲輸入(會有分詞，去掉停用詞，取詞幹等預處理)：

• 對每一個D中的文檔d，對應到不一樣topic的機率θd < pt1,..., ptk >，其中，pti表示d對應T中第i個topic的機率。計算方法是直觀的，pti=nti/n，其中nti表示d中對應第i個topic的詞的數目，n是d中全部詞的總數。

• 對每一個T中的topic t，生成不一樣單詞的機率φt < pw1,..., pwm >，其中，pwi表示t生成VOC中第i個單詞的機率。計算方法一樣很直觀，pwi=Nwi/N，其中Nwi表示對應到topic t的VOC中第i個單詞的數目，N表示全部對應到topic t的單詞總數。

LDA的核心公式以下：

p(w|d) = p(w|t)*p(t|d)

直觀的看這個公式，就是以Topic做爲中間層，能夠經過當前的θd和φt給出了文檔d中出現單詞w的機率。其中p(t|d)利用θd計算獲得，p(w|t)利用φt計算獲得。

2， RegexTokenizer

RegexTokenizer容許基於正則的方式進行文檔切分紅單詞組。默認狀況下，使用參數「pattern」（ regex, default: "s+"）做爲分隔符來分割輸入文本。或者，用戶能夠將參數「gaps」設置爲false，指示正則表達式「pattern」表示「tokens」，而不是分割間隙，並查找全部匹配事件做爲切分後的結果。

 

 

3， StopWordsRemover

stopwords簡單來講是指在一種語言中普遍使用的詞。在各類須要處理文本的地方，咱們對這些中止詞作出一些特殊處理，以方便咱們更關注在更重要的一些詞上。

中止詞的詞表通常不須要本身製做，有不少可選項能夠本身下載選用。

 

Spark中提供了StopWordsRemover類處理中止詞，它能夠用做Machine learning Pipeline的一部分。

 

StopWordsRemover的功能是直接移除全部停用詞（stopword），全部從inputCol輸入的量都會被它檢查，而後再outputCol中，這些中止詞都會去掉了。

4， CountVectorizer

CountVectorizer 和 CountVectorizerModel 旨在幫助將文本文檔集合轉化爲頻數向量。當先驗詞典不可用時，CountVectorizer能夠用做Estimator提取詞彙表，並生成一個CountVectorizerModel。該模型會基於該字典爲文檔生成稀疏矩陣，該稀疏矩陣能夠傳給其它算法，好比LDA，去作一些處理。

在擬合過程當中，CountVectorizer會從整個文檔集合中進行詞頻統計並排序後的前vocabSize個單詞。

一個可選參數minDF也會影響擬合過程，方法是指定詞彙必須出現的文檔的最小數量（或小於1.0）。另外一個可選的二進制切換參數控制輸出向量。若是設置爲true，則全部非零計數都設置爲1.這對於模擬二進制計數而不是整數計數的離散機率模型特別有用。

 

二   數據

20個主題的數據，每篇文章一個文件，每一個主題100個文件。共兩千個文件。

三  實現步驟

1， 導入數據

val corpus = sc.wholeTextFiles("file:///opt/datas/mini_newsgroups/*").map(_._2).map(_.toLowerCase())

2， 數據格式整理

val corpus_body = corpus.map(_.split("\n\n")).map(_.drop(1)).map(_.mkString(" "))

val corpus_df = corpus_body.zipWithIndex.toDF("corpus", "id")

 

import org.apache.spark.ml.feature.RegexTokenizer

val tokenizer = new RegexTokenizer().setPattern("[\W_]+").setMinTokenLength(4).setInputCol("corpus").setOutputCol("tokens")

val tokenized_df = tokenizer.transform(corpus_df)

3， 導入停用詞

val stopwords = sc.textFile("file:///opt/datas/stop_words.txt").collect()

4， 去除停用詞

import org.apache.spark.ml.feature.StopWordsRemover

 

// Set params for StopWordsRemover

val remover = new StopWordsRemover().setStopWords(stopwords).setInputCol("tokens").setOutputCol("filtered")

 

// Create new DF with Stopwords removed

val filtered_df = remover.transform(tokenized_df)

5， 生成詞頻向量

import org.apache.spark.ml.feature.CountVectorizer

 

// Set params for CountVectorizer

val vectorizer = new CountVectorizer().setInputCol("filtered").setOutputCol("features").setVocabSize(10000).setMinDF(5).fit(filtered_df)

 

val countVectors = vectorizer.transform(filtered_df).select("id", "features")

6， 構建LDA模型

import org.apache.spark.ml.clustering.LDA

 

val numTopics = 20

// Set LDA params

val lda = new LDA().setK(numTopics).setMaxIter(10)

 

7， 訓練LDA模型

val model = lda.fit(countVectors )

8， 查看訓練結果數據

val topicIndices = model.describeTopics(5)

9， 詞典的使用

val vocabList = vectorizer.vocabulary

10,使用模型

val transformed = model.transform(dataset)

transformed.show(false)

五  可調整測試點

1， 增長stop-words

val add_stopwords = Array("article", "writes", "entry", "date", "udel", "said", "tell", "think", "know", "just", "newsgroup", "line", "like", "does", "going", "make", "thanks")

val new_stopwords = stopwords.union(add_stopwords)

2， 使用EM

用於估計LDA模型的優化器或推理算法，目前Spark支持兩種：

online：Online Variational Bayes (默認)

em: Expectation-Maximization

能夠經過調用setOptimizer(value: String)，傳入online或者em來使用。