spark2.3聚类算法lda代码(python)

spark的lda有两个一个是mllib下 一个是ml下的，下面代码是使用ml的

from pyspark import SparkConf, SparkContext,SQLContext
from pyspark.sql import SparkSession 
from pyspark.ml.feature import Word2Vec,CountVectorizer
from pyspark.ml.clustering import LDA, LDAModel
from pyspark.sql.functions import col, udf
from pyspark.sql.types import IntegerType,ArrayType,StringType
import pylab as pl

def to_word(termIndices):
    words = []
    for termID in termIndices:
        words.append(vocab_broadcast.value[termID])    
    return words
conf = SparkConf().setAppName("myspark") 
sc = SparkContext(conf=conf) 
sqlContext=SQLContext(sc)

'''
documentDF = sqlContext.createDataFrame([
    ("Hi I heard about Spark".split(" "),),
    ("I wish Java could use case classes".split(" "),),
    ("Logistic regression models are neat".split(" "),)
], ["text"])
'''
#documentDF 为文档分词数据 ，格式类似上面
spark_df = sqlContext.createDataFrame(documentDF)
cv = CountVectorizer(inputCol="words", outputCol="features")
cvmodel =cv.fit(spark_df);
cvResult= cvmodel.transform(spark_df);
#10个主题 也就是10个类别
lda = LDA(k=10, maxIter=100)
ldaModel = lda.fit(cvResult)

transformed = ldaModel.transform(cvResult).select("topicDistribution")
#结果显示 每个文档各个类别的权重
transformed.show(truncate=False)
#检验上面创建lda模型中使用的参数 ll越大越好，lp越小越好
ll = ldaModel.logLikelihood(cvResult)
lp = ldaModel.logPerplexity(cvResult)

#每个主题的分词分布情况
topicIndices = ldaModel.describeTopics(maxTermsPerTopic = wordNumbers)
vocab_broadcast = sc.broadcast(vocabArray)
udf_to_word = udf(to_word, ArrayType(StringType()))

topics = topicIndices.withColumn("words", udf_to_word(topicIndices.termIndices))
topics.show(truncate=False)
exit()