如何進行SparkMllib主題模型案例的分析

如何進行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”，而不是分割間隙，并查找所有匹配事件作為切分后的結果。

具體請參考：基于DF的Tokenizer分詞

3， StopWordsRemover

stopwords簡單來說是指在一種語言中廣泛使用的詞。在各種需要處理文本的地方，我們對這些停止詞做出一些特殊處理，以方便我們更關注在更重要的一些詞上。

停止詞的詞表一般不需要自己制作，有很多可選項可以自己下載選用。

Spark中提供了StopWordsRemover類處理停止詞，它可以用作Machine learning Pipeline的一部分。

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

具體請參考，浪尖文章：基于DataFrame的StopWordsRemover處理

4， CountVectorizer

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

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

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

具體請參考，浪尖的另一篇文章：CountVectorizer

二   數據

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來使用。

看完上述內容是否對您有幫助呢？如果還想對相關知識有進一步的了解或閱讀更多相關文章，請關注億速云行業資訊頻道，感謝您對億速云的支持。