<s> from pyspark import SparkConf , SparkContext <EOL> from pyspark . mllib . clustering import LDA , LDAModel <EOL> from pyspark . mllib . linalg import Vectors <EOL> import numpy as np <EOL> def vectorizer ( sentence , vocab ) : <EOL> """<STR_LIT>""" <EOL> vocabulary = vocab . value <EOL> vec = np . zeros ( len ( vocabulary ) ) <EOL> for word in sentence . split ( "<STR_LIT:U+0020>" ) : <EOL> if word in vocabulary : <EOL> vec [ vocabulary . index ( word ) ] += <NUM_LIT:1> <EOL> return Vectors . dense ( vec ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> conf = SparkConf ( ) . setAppName ( "<STR_LIT>" ) <EOL> sc = SparkContext ( conf = conf ) <EOL> tolstoy = sc . textFile ( "<STR_LIT>" ) <EOL> sentences = tolstoy . filter ( lambda s : len ( s ) > <NUM_LIT:0> ) <EOL> word_counts = sentences . flatMap ( lambda s : s . split ( "<STR_LIT:U+0020>" ) ) . map ( lambda w : ( w . lower ( ) , <NUM_LIT:1> ) ) . reduceByKey ( lambda a , b : a + b ) <EOL> top_words = word_counts . takeOrdered ( <NUM_LIT> , key = lambda ( w , c ) : - c ) <EOL> vocabulary = [ str ( k ) for ( k , v ) in top_words ] <EOL> br_vocabulary = sc . broadcast ( vocabulary ) <EOL> dense_vectors = sentences . map ( lambda s : vectorizer ( s , br_vocabulary ) ) <EOL> corpus = dense_vectors . zipWithIndex ( ) . map ( lambda ( v , i ) : [ i , v ] ) <EOL> lda_model = LDA . train ( corpus , k = <NUM_LIT:3> , maxIterations = <NUM_LIT:20> ) <EOL> print ( "<STR_LIT>" + str ( lda_model . vocabSize ( ) ) + "<STR_LIT>" ) <EOL> topics = lda_model . topicsMatrix ( ) <EOL> for topic in range ( topics . shape [ <NUM_LIT:1> ] ) : <EOL> print ( "<STR_LIT>" + str ( topic ) + "<STR_LIT::>" ) <EOL> topic_word_counts = sorted ( zip ( vocabulary , lda_model . topicsMatrix ( ) [ : , topic ] ) , key = lambda ( w , c ) : - c ) <EOL> top_words = [ w for ( w , c ) in topic_word_counts [ : <NUM_LIT:30> ] ] <EOL> print top_words </s>
