<s> """<STR_LIT>""" <EOL> from mrjob . job import MRJob <EOL> from mrjob . protocol import JSONProtocol <EOL> from mrjob . step import MRStep <EOL> def encode_node ( node_id , links = None , score = <NUM_LIT:1> ) : <EOL> """<STR_LIT>""" <EOL> node = { } <EOL> if links is not None : <EOL> node [ '<STR_LIT>' ] = sorted ( links . items ( ) ) <EOL> node [ '<STR_LIT>' ] = score <EOL> return JSONProtocol . write ( node_id , node ) + '<STR_LIT:\n>' <EOL> class MRPageRank ( MRJob ) : <EOL> INPUT_PROTOCOL = JSONProtocol <EOL> def configure_options ( self ) : <EOL> super ( MRPageRank , self ) . configure_options ( ) <EOL> self . add_passthrough_option ( <EOL> '<STR_LIT>' , dest = '<STR_LIT>' , default = <NUM_LIT:10> , type = '<STR_LIT:int>' , <EOL> help = '<STR_LIT>' ) <EOL> self . add_passthrough_option ( <EOL> '<STR_LIT>' , dest = '<STR_LIT>' , default = <NUM_LIT> , <EOL> type = '<STR_LIT:float>' , <EOL> help = '<STR_LIT>' ) <EOL> def send_score ( self , node_id , node ) : <EOL> """<STR_LIT>""" <EOL> yield node_id , ( '<STR_LIT>' , node ) <EOL> for dest_id , weight in node . get ( '<STR_LIT>' ) or [ ] : <EOL> yield dest_id , ( '<STR_LIT>' , node [ '<STR_LIT>' ] * weight ) <EOL> def receive_score ( self , node_id , typed_values ) : <EOL> """<STR_LIT>""" <EOL> node = { } <EOL> total_score = <NUM_LIT:0> <EOL> for value_type , value in typed_values : <EOL> if value_type == '<STR_LIT>' : <EOL> node = value <EOL> else : <EOL> assert value_type == '<STR_LIT>' <EOL> total_score += value <EOL> node [ '<STR_LIT>' ] = node [ '<STR_LIT>' ] <EOL> d = self . options . damping_factor <EOL> node [ '<STR_LIT>' ] = <NUM_LIT:1> - d + d * total_score <EOL> yield node_id , node <EOL> def steps ( self ) : <EOL> return ( [ MRStep ( mapper = self . send_score , reducer = self . receive_score ) ] * <EOL> self . options . iterations ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> MRPageRank . run ( ) </s>
