<s> """<STR_LIT>""" <EOL> from mrjob . job import MRJob <EOL> from mrjob . step import MRStep <EOL> import re <EOL> WORD_RE = re . compile ( r"<STR_LIT>" ) <EOL> class MRNextWordStats ( MRJob ) : <EOL> SORT_VALUES = True <EOL> def steps ( self ) : <EOL> return [ MRStep ( mapper = self . m_find_words , <EOL> combiner = self . c_combine_counts , <EOL> reducer = self . r_sum_counts ) , <EOL> MRStep ( reducer = self . r_compute_stats ) ] <EOL> def m_find_words ( self , _ , line ) : <EOL> """<STR_LIT>""" <EOL> prev_word = None <EOL> for word in WORD_RE . findall ( line ) : <EOL> word = word . lower ( ) <EOL> if prev_word is not None : <EOL> yield ( prev_word , '<STR_LIT:*>' ) , <NUM_LIT:1> <EOL> yield ( prev_word , word ) , <NUM_LIT:1> <EOL> prev_word = word <EOL> def c_combine_counts ( self , key , counts ) : <EOL> """<STR_LIT>""" <EOL> yield key , sum ( counts ) <EOL> def r_sum_counts ( self , key , counts ) : <EOL> """<STR_LIT>""" <EOL> count = sum ( counts ) <EOL> prev_word , word = key <EOL> if word == '<STR_LIT:*>' : <EOL> yield prev_word , ( '<STR_LIT>' , count ) <EOL> else : <EOL> yield prev_word , ( '<STR_LIT>' , ( word , count ) ) <EOL> def r_compute_stats ( self , prev_word , value ) : <EOL> """<STR_LIT>""" <EOL> total = None <EOL> for value_type , data in value : <EOL> if value_type == '<STR_LIT>' : <EOL> total = data <EOL> else : <EOL> assert value_type == '<STR_LIT>' <EOL> word , count = data <EOL> percent = <NUM_LIT> * count / total <EOL> yield ( prev_word , word ) , ( total , count , percent ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> MRNextWordStats . run ( ) </s>
