<s> """<STR_LIT>""" <EOL> import re <EOL> import math <EOL> from operator import add <EOL> from functools import partial <EOL> from pyspark import SparkConf , SparkContext <EOL> APP_NAME = "<STR_LIT>" <EOL> N_DOCS = <NUM_LIT> <EOL> header = re . compile ( r'<STR_LIT>' , re . I ) <EOL> tokenizer = re . compile ( r'<STR_LIT>' ) <EOL> def chunk ( args ) : <EOL> """<STR_LIT>""" <EOL> def chunker ( text ) : <EOL> """<STR_LIT>""" <EOL> docid = None <EOL> document = [ ] <EOL> for line in text . split ( "<STR_LIT:\n>" ) : <EOL> hdr = header . match ( line ) <EOL> if hdr is not None : <EOL> if docid is not None : <EOL> yield ( docid , document ) <EOL> docid = hdr . groups ( ) [ <NUM_LIT:0> ] <EOL> document = [ ] <EOL> continue <EOL> else : <EOL> document . append ( line ) <EOL> fname , text = args <EOL> return list ( chunker ( text ) ) <EOL> def tokenize ( document , stopwords = None ) : <EOL> """<STR_LIT>""" <EOL> def inner_tokenizer ( lines ) : <EOL> """<STR_LIT>""" <EOL> for line in lines : <EOL> for word in re . split ( tokenizer , line ) : <EOL> if word and word not in stopwords . value and word . isalpha ( ) : <EOL> yield word <EOL> docid , lines = document <EOL> return [ <EOL> ( ( docid , word ) , <NUM_LIT:1> ) for word in inner_tokenizer ( lines ) <EOL> ] <EOL> def term_frequency ( v1 , v2 ) : <EOL> """<STR_LIT>""" <EOL> docid , tf , count1 = v1 <EOL> _docid , _tf , count2 = v2 <EOL> return ( docid , tf , count1 + count2 ) <EOL> def tfidf ( args ) : <EOL> """<STR_LIT>""" <EOL> ( key , ( tf , n ) ) = args <EOL> if n > <NUM_LIT:0> : <EOL> idf = math . log ( N_DOCS / n ) <EOL> return ( key , idf * tf ) <EOL> def main ( sc ) : <EOL> """<STR_LIT>""" <EOL> with open ( '<STR_LIT>' , '<STR_LIT:r>' ) as words : <EOL> stopwords = frozenset ( [ <EOL> word . strip ( ) for word in words . read ( ) . split ( "<STR_LIT:\n>" ) <EOL> ] ) <EOL> stopwords = sc . broadcast ( stopwords ) <EOL> corpus = sc . wholeTextFiles ( '<STR_LIT>' ) . flatMap ( chunk ) <EOL> docfreq = corpus . flatMap ( partial ( tokenize , stopwords = stopwords ) ) <EOL> docfreq = docfreq . reduceByKey ( add ) <EOL> trmfreq = docfreq . map ( lambda ( key , tf ) : ( key [ <NUM_LIT:1> ] , ( key [ <NUM_LIT:0> ] , tf , <NUM_LIT:1> ) ) ) <EOL> trmfreq = trmfreq . reduceByKey ( term_frequency ) <EOL> trmfreq = trmfreq . map ( lambda ( word , ( docid , tf , n ) ) : ( ( word , docid ) , ( tf , n ) ) ) <EOL> tfidfs = trmfreq . map ( tfidf ) <EOL> tfidfs . saveAsTextFile ( "<STR_LIT>" ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> conf = SparkConf ( ) . setAppName ( APP_NAME ) <EOL> sc = SparkContext ( conf = conf ) <EOL> main ( sc ) </s>
