Pyspark聚类--LDA_pyspark lda

LDA

class pyspark.ml.clustering.LDA(featuresCol=‘features’, maxIter=20, seed=None, checkpointInterval=10, k=10, optimizer=‘online’, learningOffset=1024.0, learningDecay=0.51, subsamplingRate=0.05, optimizeDocConcentration=True, docConcentration=None, topicConcentration=None, topicDistributionCol=‘topicDistribution’, keepLastCheckpoint=True)

潜在狄利克雷分配 (LDA)，一种专为文本文档设计的主题模型。

术语：

​ “term” = “word”：词汇的一个元素

​ “token”：出现在文档中的术语实例

​ “topic”：代表某个概念的术语的多项分布

​ “document”：一段文本，对应输入数据中的一行

原始 LDA 论文（期刊版）：

​ Blei, Ng, and Jordan. “Latent Dirichlet Allocation.” JMLR, 2003.

输入数据（featuresCol）：通过 featuresCol 参数给 LDA 一个文档集合作为输入数据。每个文档都被指定为Vector长度为 vocabSize，其中每个条目是文档中相应术语（单词）的计数。诸如 pyspark.ml.feature.Tokenizer和之类的特征转换pyspark.ml.feature.CountVectorizer器可用于将文本转换为字数向量。

&&&&& LDAmodel有单机和分布式两种类型，详细描述见pyspark官方文档

checkpointInterval = Param(parent=‘undefined’, name=‘checkpointInterval’, doc=‘设置检查点间隔 (>= 1) 或 disable checkpoint (-1)。例如 10 表示缓存将每 10 次迭代获得检查点。注意： 如果检查点目录未在 SparkContext 中设置，则此设置将被忽略。’）

docConcentration = Param(parent=‘undefined’, name=‘docConcentration’, doc=‘优先放置在文档上的浓度参数（通常称为“alpha”）在主题上的分布（“theta”）。’)

k = Param(parent=‘undefined’, name=‘k’, doc=‘要推断的主题（簇）的数量。必须> 1。’)

keepLastCheckpoint = Param(parent=‘undefined’, name=‘keepLastCheckpoint ‘, doc=’(对于 EM 优化器) 如果使用检查点，则表示是否保留最后一个检查点。如果为 false，则检查点将被删除。如果数据分区丢失，删除检查点会导致失败，因此小心设置。’)

learningDecay = Param(parent=‘undefined’, name=‘learningDecay’, doc=‘学习率，设置为指数衰减率。这应该在 (0.5, 1.0] 之间以保证渐近收敛。’)

learningOffset = Param(parent=‘undefined’, name=‘learningOffset’, doc=‘降低早期迭代权重的（正）学习参数。较大的值使早期迭代计数更少’）

subsamplingRate = Param(parent=‘undefined’, name=‘subsamplingRate’, doc=‘小批量梯度下降每次迭代中要采样和使用的语料的分数，范围为 (0, 1].’)

topicConcentration = Param(parent=‘undefined’, name=‘topicConcentration’, doc=‘浓度参数（通常称为“beta”或“eta”），用于优先放置在 topic’ 上的术语分布。’)

topicDistributionCol = Param(parent =‘undefined’, name=‘topicDistributionCol’, doc=‘输出列，估计每个文档的主题混合分布（在文献中通常称为“theta”）。返回一个空文档的零向量。’）

model.describeTopics(maxTermsPerTopic=10)：返回由权重最高的术语描述的主题。

model.logLikelihood(dataset)：计算整个语料库的对数似然的下限。 参见在线 LDA 论文（Hoffman 等人，2010 年）中的公式 (16)。 警告：如果这个模型是 DistributedLDAModel 的一个实例（当优化器设置为“em”时产生），这涉及到向驱动程序收集一个大的 topicsMatrix()。 将来可能会更改此实现。

model.logPerplexity(dataset)：计算困惑度的上限。（越低越好。）参见在线 LDA 论文中的公式 (16) (Hoffman et al., 2010)。

警告：如果这个模型是 DistributedLDAModel 的一个实例（当优化器设置为“em”时产生），这涉及到向驱动程序收集一个大的topicsMatrix()。 将来可能会更改此实现。

model.topicsMatrix()：推断的主题，其中每个主题由术语的分布表示。 这是一个大小为 vocabSize x k 的矩阵，其中每一列都是一个主题。 不保证主题的顺序。 警告：如果此模型实际上是由期望最大化（“em”）优化器生成的 DistributedLDAModel 实例，则此方法可能涉及向驱动程序收集大量数据（按 vocabSize x k 的顺序）。

model.isDistributed().：指示此实例是否为 DistributedLDAModel (分布式)类型

model.vocabSize()：词汇量（词汇中术语或单词的数量

01.创建数据：

from pyspark.sql import SparkSession
spark = SparkSession.builder.config("spark.driver.host","192.168.1.10")\
    .config("spark.ui.showConsoleProgress","false").appName("LDA")\
    .master("local[*]").getOrCreate()
from pyspark.ml.linalg import Vectors, SparseVector
df = spark.createDataFrame([
    [1, Vectors.dense([0.0, 1.0])],
    [2, SparseVector(2, {0: 1.0})],
], ["id", "features"])
df.show()
df.printSchema()
1
2
3
4
5
6
7
8
9
10
11

​ 输出结果：

+---+-------------+
| id|     features|
+---+-------------+
|  1|    [0.0,1.0]|
|  2|(2,[0],[1.0])|
+---+-------------+

root
 |-- id: long (nullable = true)
 |-- features: vector (nullable = true)
1
2
3
4
5
6
7
8
9
10

02.构建LDA聚类模型，并转换原始数据进行查看

from pyspark.ml.clustering import LDA
lda = LDA(k=2, seed=1, optimizer="em")
model = lda.fit(df)
model.transform(df).show()
print(model.transform(df).head(2))
1
2
3
4
5

​ 输出结果：

+---+-------------+--------------------+
| id|     features|   topicDistribution|
+---+-------------+--------------------+
|  1|    [0.0,1.0]|[0.50000460701739...|
|  2|(2,[0],[1.0])|[0.49999493141516...|
+---+-------------+--------------------+

[Row(id=1, features=DenseVector([0.0, 1.0]), topicDistribution=DenseVector([0.5, 0.5])), 
 Row(id=2, features=SparseVector(2, {0: 1.0}), topicDistribution=DenseVector([0.5, 0.5]))]
1
2
3
4
5
6
7
8
9

03.查看是否时分布式模型，如果不是就转为单机local模型

print(model.isDistributed())
localModel = model.toLocal()
print(localModel.isDistributed())
1
2
3

​ 输出结果：

True
False
1
2

04.查看词汇量

print(model.vocabSize())
1

​ 输出结果：

2
1

05.查看推断主题矩阵

print(model.topicsMatrix())
1

​ 输出结果：

DenseMatrix([[0.49989808, 0.50010192],
             [0.50010192, 0.49989808]])
1
2

06.查看主题描述

model.describeTopics().show()
print(model.describeTopics().head(2))
1
2

​ 输出结果：

+-----+-----------+--------------------+
|topic|termIndices|         termWeights|
+-----+-----------+--------------------+
|    0|     [1, 0]|[0.50010191915681...|
|    1|     [0, 1]|[0.50010191915681...|
+-----+-----------+--------------------+

[Row(topic=0, termIndices=[1, 0], termWeights=[0.500101919156814, 0.499898080843186]), 
Row(topic=1, termIndices=[0, 1], termWeights=[0.500101919156814, 0.499898080843186])]
1
2
3
4
5
6
7
8
9