sparkmllib算法实例(sparkmllib系统推荐)

很多新手对于如何分析SparkMllib主题模型案例不是很清楚。为了帮助大家解决这个问题，下面小编就详细讲解一下。需要的人可以从中学习，希望你能有所收获。

一 文章涉及到的算法

1， LDA主题模型

符号定义

集合d，m个文档，主题集合t，k个主题

在D中，每个文档D被视为一个单词序列w1、w2，wn，wi代表第I个单词，让D有n个单词。(在LDA中称为词包，其实每个词的位置对LDA算法没有影响)

D中涉及的所有不同的词形成了一大组呼语(简称VOC)

LDA符合的分布

每篇文章D(长度为)都有自己的主题分布，主题分布是多项式分布。这个多项式分布的参数服从狄利克雷分布，这个狄利克雷分布的参数是。

每个主题都有自己的单词分布。字分布是多项式分布，这个多项式分布的参数服从狄利克雷分布，这个狄利克雷分布的参数是。

对于文章中的第n个单词，首先从文章的主题分布中抽取一个主题，然后从该主题对应的单词分布中抽取一个单词。重复这个随机生成过程，直到所有m篇文章都完成。

结果是训练两个结果向量(k个主题，VOC包含m个单词)

LDA以文档集D为输入(会有分词、停词、词干提取等预处理。):

对于每个D中的文档D，概率d pt1，ptk对应不同话题，其中pti表示D对应t中第ith个话题的概率，计算方法直观，pti=nti/n，其中nti表示D中第ith个话题对应的字数，n为D中所有字的总数。

对于每个t中的主题t，生成概率t pw1，不同字的pwm，其中pwi表示t在VOC中生成第ith个字的概率。计算方法也很直观，pwi=Nwi/N，其中Nwi表示对应话题t的VOC中第ith个单词的个数，N表示对应话题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:)用作分隔符来分割输入文本。或者，用户可以将参数“gap”设置为false，表示正则表达式“pattern”的意思是“tokens”而不是拆分gap，并查找所有匹配的事件作为拆分结果。

请参考：基于DF的分词器分词

3， StopWordsRemover

单词只是指一种语言中广泛使用的非单词。在我们需要处理文本的各个地方，我们对这些停止词进行一些特殊的处理，以便我们可以更多地关注一些更重要的词。

Stopword的词汇一般不需要自己制作，有很多选项可以自己下载选择。

Spark中提供了StopWordsRemover类来处理stopword，它可以用作机器学习管道的一部分。

去字功能是直接去掉所有的停止字(stop word)，它会检查inputCol的所有输入量，然后在outputCol中，这些停止字会被去掉。

有关详细信息，请参考波前文章：基于数据帧的停止字去除器处理

4， CountVectorizer

CountVectorizer和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来使用。

看完上述内容是否对您有帮助呢？如果还想对相关知识有进一步的了解或阅读更多相关文章，请关注行业资讯频道，感谢您对的支持。