本文次要在 Spark 平台下实现一个机器学习利用，该利用次要波及 LDA 主题模型以及 K -means 聚类。通过本文你能够理解到：
- 文本开掘的根本流程
- LDA 主题模型算法
- K-means 算法
- Spark 平台下 LDA 主题模型实现
- Spark 平台下基于 LDA 的 K -means 算法实现
1. 文本开掘模块设计
1.1 文本开掘流程
文本剖析是机器学习中的一个很宽泛的畛域，并且在情感剖析、聊天机器人、垃圾邮件检测、举荐零碎以及自然语言解决等方面失去了广泛应用。
文本聚类是信息检索畛域的一个重要概念，在文本开掘畛域有着宽泛的利用。文本聚类可能主动地将文本数据集划分为不同的类簇，从而更好地组织文本信息，能够实现高效的常识导航与浏览。
本文抉择主题模型 LDA(Latent Dirichlet Allocation)算法对文档进行分类解决，抉择在 Spark 平台上通过 Spark MLlib 实现 LDA 算法，其中 Spark Mllib 是 Spark 提供的机器学习库，该库提供了罕用的机器学习算法。其根本设计思路如下图所示：
1.2 文本开掘流程剖析
首先是数据源局部，次要的数据包含文档数据和互联网爬虫数据。而后是数据抽取局部，将曾经收集好的数据通过同步工具上传至分布式文件系统 HDFS，作为模型训练的数据源。其次是数据摸索与预处理局部，该局部次要是对原始数据集进行预处理，包含分词、停用词过滤、特征向量提取等。再次是模型训练局部，次要包含训练与测试，从而失去一个模型。最初是模型评估，对学得模型进行评估之后，进行线上部署。
2. 文本开掘模块算法钻研
2.1LDA 主题模型算法
LDA（Latent Dirichlet allocation）由 David M. Blei，Andrew Y. Ng，Michael I. Jordan 于 2003 年提出的基于概率模型的主题模型算法，即隐含狄利克雷散布，它能够将文档集中每篇文档的主题以概率分布的模式给出，将文本向量投射到更容易剖析解决的主题空间当中，去除文本中存在的噪声，是一种罕用的文本剖析技术，能够用来辨认大规模文档集或语料库中潜在的主题信息，通常被用来对大规模文档数据进行建模。通过主题模型和设定的主题数，能够训练出文档汇合中不同的主题所占的比例以及每个主题下的要害词语呈现的概率。从文档汇合中学习失去主题散布和主题比例，能够进一步在数据挖掘工作中应用。
LDA 借用词袋的思维，以某一概率选取某个主题，再以某一概率选出主题中的每个单词，通过一直反复该步骤产生文档中的所有语词。该办法对词汇进行了含糊聚类，汇集到一类的词能够间接地示意一个隐含的主题。LDA 对文本信息进行了开掘，能用来掂量不同文档之间的潜在关系，也能通过某一类词来表白文档中暗藏的主题。
2.2K 均值算法
聚类 (Clustering) 是一种将数据集划分为若干组或类的办法。通过聚类过程将一群形象的对象分为若干组，每一组由类似的对象形成，称之为一个类别。与分类不同(将数据依照当时定义好的分类规范进行划分)，聚类是一种无监督学习(unsupervised learning)，训练数据集的标签信息是未知的，指标是通过对无标记训练样本依照特定的测度的形似性水平进行聚合，为进一步数据分析提供根底。
K 均值 (k-means) 算法的根本思维是初始随机给定 K 个簇核心，即从 n 个数据对象中抉择 k 个任意对象作为初始的簇核心，依照最邻近准则把待分类样本点分到各个簇。而后按平均法从新计算各个簇的核心(该类别中的所有数据对象的均值)，从而确定新的簇心。始终迭代，直到簇心的挪动间隔小于某个给定的值。
K 均值算法采纳了贪婪策略，通过迭代优化来近似求解上式 E 值，算法流程如下图所示
2.3 文本开掘算法优化
LDA 主题模型算法利用于文档聚类，计算得出的主题能够看做是文档的聚类核心，利用主题模型进行文档聚类，能够无效地组织文档数据集。同时，因为 LDA 主题模型能够计算出每篇文档在不同主题下的概率分布，因而能够将此主题的概率分布作为文档的特征向量，从而将高维的文档向量投影到低维的特色空间中。
计算文本之间的间隔是传统的 K -means 算法在进行文本聚类时的关键步骤，而文本通常是非结构化数据，构建的文本向量具备稠密性和维度高的特点，同时，构建文本特征向量并未思考到文字之间的语义关系，因而可能会造成位于同一类簇的文本之间具备非相似性。
因而本文基于 LDA 主题模型改良 K -means 算法，首先通过 LDA 主题模型对文档数据集进行建模，挖掘出每篇文档的主题概率分布，既可能达到文档降维和去除噪声的成果，又能补救通过关键词构建文档特征向量容易造成失落信息的缺点。最初每篇文档的主题概率分布作为 K -means 算法的输出数据集。
3. 试验剖析
3.1 基于 Spark 的 LDA 主题模型算法实现
数据集介绍
抉择 Newsgroups 数据集作为该试验的训练集和测试集。Newgroups 是一个新闻数据集，该数据集包含大概 20000 篇新闻文档，总共分为 6 个大类别，每个大类别又分不同的小类别，小类别共计 20 个，如下表所示。该新闻数据集曾经成为了学界和业界在机器学习的文本开掘试验中罕用的数据集，比方文本分类和文本聚类。
该数据集共蕴含 7 个文件，其中 3 个文件为训练数据(train.data、train.label、train.map)，共计 11269 篇，另外 3 个文件为测试数据(test.data、test.label、test.map)，共计 7505 篇，另外一个文件为词汇表(vocabulary.txt), 其第 i 行示意编号为 i 的单词的名称。文件扩大名为.data 的文件格式为[docIdx wordIdx count]，其中 docIdx 示意文档编号，wordIdx 示意词语的编号，count 示意该词语的词频统计。文件扩大名为.label 的文件示意文档的主题分类，每行数据代表某篇文档的类别。文件扩大名为.map 的文示意类别编号与类别名称的映射关系，其具体格局为[labelName labelId]。
原始数据集解决
原始的数据集格局为 [docIdx wordIdx count]，例如[1,20,2] 示意在编号为 1 的文档中，编号为 20 的词语的词频是 2。LDA 承受的参数格局为：
[label,(vector_ size, [wiIdx,wjIdx,···wnIdx],[tfi,tfj,···tfn])]上述格局的数据代表一个带有标签的稠密向量，其中 label 示意文档编号，vector_ size 示意向量的维度，wnIdx 示意词 n 的索引编号，tfn 示意词 n 的词频。须要将原始数据转换成上述的格局，具体步骤如下：
- Step1：将原始数据集上传至 HDFS
[kms@kms-1 ~]$ hdfs dfs -put /opt/modules/train_data/lda/train.data /train/lda/- Step2: 初始化 SparkSession 并加载数据
val spark = SparkSession
.builder
.appName(s"${this.getClass.getSimpleName}") .getOrCreate()
// 设置日志级别
Logger.getLogger("org.apache.spark").setLevel(Level.OFF) Logger.getLogger("org.apache.hadoop").setLevel(Level.OFF)
// 加载原始数据集
val rowDS = spark.read.textFile("/train/lda/train.data") - Step3: 数据集矩阵变换处理
// 创立形如 MatrixEntry(row_index, column_index, value)的 MatrixEntry
val matrixEntry:RDD[MatrixEntry] = rowDS.rdd.map(_.split(" "))
.map(rowdata => MatrixEntry(rowdata(0).toLong,rowdata(1).toLong,rowdata(2).toDouble))
// 创立稠密矩阵
val sparseMatrix: CoordinateMatrix = new CoordinateMatrix(matrixEntry)
// 创立 LabeledPoint 数据集
val labelPointData = sparseMatrix.toIndexedRowMatrix.rows.map(r => (r.index, r.vector.asML))
val corpusDF = spark.createDataFrame(labelPointData).toDF("label","features")
corpusDF.saveAsTextFile("/tarin/lda/labelPointData")
解决之后的局部数据集如下所示，其中一行代表一篇文档的特征向量
[4551,(53976,[23,27,29,30,44,45,48,314,425,748,767,825,930,969,995,1345,7033,13872,16798,19139,26846,26847,27081,29607,30801,31200,31201,31202],[2.0,1.0,3.0,3.0,1.0,1.0,1.0,1.0,2.0,3.0,1.0,2.0,1.0,1.0,2.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0])]
[2493,(53976,[80,133,754,3699,4066,5190,6138,7327,7361,10267,10344,10949,11390,11683,11759,16206,22708,22709],[1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,2.0,1.0,1.0,2.0,1.0,1.0,1.0,1.0,2.0,1.0])]
k 折穿插验证确定训练参数
穿插验证法 (cross validation) 是将数据集 D 划分为 k 个大小类似的互斥子集的一种办法，其中每个子集都尽可能地放弃数据分布的一致性，即从数据集 D 中通过分层采样的形式失去。而后，每次再用 k - 1 个子集的并集作为训练集，剩下的那个子集作为测试集；通过这样的解决，能够失去 k 组训练集和测试集，进而能够进行 k 次训练和测试，最终返回的是这 k 个测试后果的均值。穿插验证法评估后果的稳定性和保真性在很大水平上依赖于 k 的取值，为突出这一点，通常把穿插验证法称为 k 折穿插验证(k-fold cross validation)。K 通常取值为 10，称之为 10 折穿插验证, 下图给出了 10 折穿插验证的示意图。
困惑度 (perplexity) 指标是 LDA 模型的原作者 Blei 等提出的一种反馈模型泛化能力的指标, 在评估模型的优劣上具备肯定的代表性和普遍性。所谓困惑度就是文档在划分主题时确定性的评判, 反映的是模型对新样本的适用性。
十折穿插验证处理过程如下所示
// 将数据集宰割为 10 份，每份占 10%
val splitData = labelPointData.randomSplit(Array.fill(10)(0.1))
// 设定主题的个数为 15-25
val rangTopic = 15 to 25
rangTopic.foreach { k =>
var perplexity = 0.0
for (i <- 0 to 9) {
// 抉择其中 9 份做训练集
val trainIdx = (0 to 9).toArray.filter(_ != i)
var trainData = spark.sparkContext.union(splitData(trainIdx(0)),
splitData(trainIdx(1)),
splitData(trainIdx(2)),
splitData(trainIdx(3)),
splitData(trainIdx(4)),
splitData(trainIdx(5)),
splitData(trainIdx(6)),
splitData(trainIdx(7)),
splitData(trainIdx(8)))
// 创立 DataFrame
val trainDF = spark.createDataFrame(trainData).toDF("label","features")
val testDF = spark.createDataFrame(splitData(i)).toDF("label","features")
// 训练主题个数为 k 时的模型
val lda = new LDA().setK(k).setMaxIter(50)
val ldaModel = lda.fit(trainDF)
perplexity = perplexity + ldaModel.logPerplexity(testDF)
}
val avePerplexity = perplexity / 10
System.out.println("当主题个数为" + k + "时，" + "穿插验证的均匀困惑度为" + avePerplexity)
}
通过十折穿插验证，验证在不同主题下 (取值 15-25) 训练模型的均匀困惑度，测试发现在主题 k =20 时，困惑度的值最小。因为困惑度值越小, 示意该模型具备较好的泛化能力，所以抉择 k =20 作为主题个数。
试验后果
将主题个数设置为 20，迭代次数设置为 50，应用上述数据集训练 LDA 模型，具体步骤如下
// 训练 LDA 模型
val lda = new LDA().setK(20).setMaxIter(50)
val model = lda.fit(corpusDF)
val ll = model.logLikelihood(corpusDF)
val lp = model.logPerplexity(corpusDF)
println("当主题个数为 20 时对数似然为:" + ll)
println("当主题个数为 20 时困惑度为:" + lp)
// 形容主题
val topics = model.describeTopics(5)
println("The topics described by their top-weighted terms:")
topics.show(false)
topics.rdd.saveAsTextFile("/tarin/lda/topics")
// 测试后果
val transformed = model.transform(corpusDF)
transformed.select("label","topicDistribution").rdd.saveAsTextFile("/tarin/lda /testtopic")
通过训练失去 LDA 模型，其中训练数据的主题 - 单词概率分布如下表所示，抉择权重排名在前 5 的单词，其中 topic 示意主题编号，termIndices 示意词语编号组成的汇合，termWeights 示意词语编号对应的权重的汇合。
每个主题对应的单词列表如下表所示，其中 topic 示意主题，termIndices 示意词语编号组成的汇合,vocabulary 示意词汇。
该模型的文档 - 主题散布如下表所示，因为文档较多，这里仅列出局部文档，其中 label 示意文档编号，topicDistribution 示意文档的主题散布。
通过下面的剖析，失去了文本的主题散布。每篇文档将对应一个主题的特色空间，从而达到降维的目标。主题 - 单词单词概率分布形容了主题的特色空间，其中主题示意聚类核心。
联合词汇表与训练集，将其解决成 [word,count] 的模式，其中 word 示意单词，count 示意该次呈现的频次，具体的处理过程如下。
package com.apache.ml
import org.apache.spark.{SparkConf, SparkContext}
object WordCount {def main(args: Array[String]): Unit = {val conf = new SparkConf()
.setMaster("local")
.setAppName("wordcount")
val sc = new SparkContext(conf)
val rawRDD = sc.textFile("e:///lda/train.data")
val vocabulary = sc.textFile("e:///lda/vocabulary.txt")
val word_nums = rawRDD.map(_.split(" ")).map(row => (row(1).toLong, row(2).toInt))
val word_count = word_nums.reduceByKey(_ + _)
val sort_wcnt = word_count.sortByKey(false)
// 解决词汇表
val num_vocabulary = vocabulary.zipWithIndex().map(row => (row._2, row._1))
val sort_combination = sort_wcnt.join(num_vocabulary)
.map(row => (row._2._1, row._2._2))
.sortByKey(false)
.map(row => (row._2, row._1))
sort_combination.saveAsTextFile("e:///combination")
}
}
通过应用 R 语言的 wordcloud2 包，进行可视化文档词云图展现，见下图
3.2Spark 平台下基于 LDA 的 k -means 算法实现
数据预处理
将通过 LDA 主题模型计算的文档 - 主题散布作为 k -means 的输出，文档 - 主题散布的模式为 [label, features，topicDistribution]，其中 features 代表文档的特征向量，每一行数据代表一篇文档。因为 k -means 承受的特征向量输出的模式为[label，features]，所以须要将原始的数据集 schema 转化为[label，features] 的模式，行将 topicDistribution 列名转为 features。解决步骤为：
Val trainDF = transformed.select("label","topicDistribution").toDF("label", "features") 模型训练
Spark ML 的 K -means 算法提供了如下的参数配置：
- setFeaturesCol(value: String)：设置输出的特征向量列，默认值为 features
- setK(value: Int)：设置类簇的个数
- setMaxIter(value: Int)：设置最大迭代次数
- setPredictionCol(value: String)：设置输入列名称，默认为 prediction
- setSeed(value: Long)：设置随机数种子
- setTol(value: Double)：设置收敛阈值
设置最大迭代次数为 200，随机数种子为 123，类簇个数为 2、4、6、8、10、12、14、16、18、20，其余抉择默认值，别离察看评估指标的变动状况。具体代码如下：
(2 to 20 by 2).toList.map {k =>
val kmeans = new KMeans().setK(k).setSeed(123).setMaxIter(200)
val kmeansModel = kmeans.fit(kMeansTrain)
// 预测后果
val predictions = kmeansModel.transform(kMeansTrain)
// 计算误差平方和
val wssse = kmeansModel.computeCost(kMeansTrain)
println(s"Within set sum of squared errors = $wssse")
// 计算轮廓系数
val evaluator = new ClusteringEvaluator()
val silhouette = evaluator.evaluate(predictions)
println(s"Silhouette with squared euclidean distance = $silhouette")
// 显示聚类后果
println("Cluster Centers:")
kmeansModel.clusterCenters.foreach(println)
}经过训练，失去当 K 为 6 时聚类成果最好，此时的聚类后果如下表所示：
模型评估
- 轮廓系数
轮廓系数（Silhouette Coefficient）是评估聚类成果好坏的一种办法，用来掂量簇的密集与扩散水平。它联合内聚度和拆散度两种因素，能够用来在雷同原始数据的根底上用来评估不同算法、或者算法不同运行形式对聚类后果所产生的影响。轮廓系数取值为[-1,1], 其值越大示意同类中样本间隔最近，不同类中样本间隔最远，即该值越靠近于 1，簇越紧凑，聚类成果越好。
应用 K -means 算法，将待分类数据集分为了 k 个类簇，对于其中的一个点 i 来说，a(i)示意该向量到它所属类簇中其余点的均匀间隔，b(i)示意该向量到其余类簇中点的均匀间隔。对于一个样本汇合，所有样本的轮廓系数的平均值即为该聚类后果总的轮廓系数。
- 误差平方和
误差平方和又称残差平方和、组内平方和等。依据 n 个察看值拟合适当的模型后，余下未能拟合部份 (ei=yi- y 均匀) 称为残差，其中 y 均匀示意 n 个察看值的平均值，所有 n 个残差平方之和称误差平方和。
当 K 取不同值时，计算所得误差平方和如下所示：
计算所得的轮廓系数如下图所示，联合误差平方和和轮廓系数，当 k = 6 时，有着较好的聚类成果。
3.3 后果剖析
首先，通过 LDA 主题模型，能够计算出文档数据集的文档 - 主题散布状况和主题 - 单词的散布状况，训练得出的主题数即为类簇数。
对 LDA 训练的文档 - 主题散布后果，行将文档示意成在不同主题上的散布所组成的向量，因为 LDA 思考到了词之间的语义关系，所以该特征向量可能更好地反馈文档的信息，因而能够将其作为 K -means 聚类算法的输出，从而补救基于空间向量模型的 K -means 算法的毛病。通过试验发现，在类簇 K 为 6 时，轮廓系数为 65.9661577458792，误差平方和为 0.8266340036962969，聚类成果良好。
小结
本章次要介绍 Spark 平台下基于 LDA 的 k -means 算法实现。对文本开掘进行了具体设计，在公开数据集上训练 LDA 模型，并对文档 - 主题散布和主题 - 词语散布进行了具体阐明。最初实现了基于 LDA 的 K -means 聚类算法，克服了传统 K -means 算法的缺点。
公众号『大数据技术与数仓』，回复『材料』支付大数据资料包