<s> import sys , os . path <EOL> sys . path . insert ( <NUM_LIT:0> , os . path . dirname ( os . path . dirname ( os . path . abspath ( __file__ ) ) ) ) <EOL> from dpark import DparkContext <EOL> from dpark . bagel import Vertex , Edge , Bagel <EOL> def parse_vertex ( line , numV ) : <EOL> fields = line . split ( '<STR_LIT:U+0020>' ) <EOL> title , refs = fields [ <NUM_LIT:0> ] , fields [ <NUM_LIT:1> : ] <EOL> outEdges = [ Edge ( ref ) for ref in refs ] <EOL> return ( title , Vertex ( title , <NUM_LIT:1.0> / numV , outEdges , True ) ) <EOL> def gen_compute ( num , epsilon ) : <EOL> def compute ( self , messageSum , agg , superstep ) : <EOL> if messageSum and messageSum [ <NUM_LIT:0> ] : <EOL> newValue = <NUM_LIT> / num + <NUM_LIT> * messageSum [ <NUM_LIT:0> ] <EOL> else : <EOL> newValue = self . value <EOL> terminate = ( superstep >= <NUM_LIT:10> and abs ( newValue - self . value ) < epsilon ) or superstep > <NUM_LIT:30> <EOL> outbox = [ ( edge . target_id , newValue / len ( self . outEdges ) ) <EOL> for edge in self . outEdges ] if not terminate else [ ] <EOL> return Vertex ( self . id , newValue , self . outEdges , not terminate ) , outbox <EOL> return compute <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> inputFile = '<STR_LIT>' <EOL> threshold = <NUM_LIT> <EOL> dpark = DparkContext ( ) <EOL> path = os . path . join ( os . path . dirname ( os . path . abspath ( __file__ ) ) , inputFile ) <EOL> input = dpark . textFile ( path ) <EOL> numVertex = input . count ( ) <EOL> vertices = input . map ( lambda line : parse_vertex ( line , numVertex ) ) . cache ( ) <EOL> epsilon = <NUM_LIT> / numVertex <EOL> messages = dpark . parallelize ( [ ] ) <EOL> result = Bagel . run ( dpark , vertices , messages , <EOL> gen_compute ( numVertex , epsilon ) ) <EOL> for id , v in result . filter ( lambda ( id , v ) : v . value > threshold ) . collect ( ) : <EOL> print id , v </s>
