<s> import sys <EOL> sys . path . append ( '<STR_LIT>' ) <EOL> import logging <EOL> import dpark <EOL> name = '<STR_LIT>' <EOL> def parse ( line ) : <EOL> sid , uid , r , f = line . strip ( ) . split ( '<STR_LIT:\t>' ) <EOL> defaults = { '<STR_LIT:F>' : <NUM_LIT> , '<STR_LIT:P>' : <NUM_LIT> , '<STR_LIT:N>' : <NUM_LIT> } <EOL> if r == '<STR_LIT:None>' : <EOL> r = defaults [ f ] <EOL> return ( sid , ( uid , float ( r ) ) ) <EOL> rating = dpark . textFile ( name , numSplits = <NUM_LIT:2> ) . map ( parse ) . groupByKey ( <NUM_LIT:2> ) <EOL> print rating . count ( ) <EOL> def reverse ( it ) : <EOL> s = { } <EOL> for k , us in it : <EOL> for u , r in us : <EOL> s . setdefault ( u , { } ) [ k ] = r <EOL> return s <EOL> def vsum ( a , b ) : <EOL> if len ( a ) < len ( b ) : <EOL> a , b = b , a <EOL> d = dict ( a ) <EOL> s = <NUM_LIT:0> <EOL> for u , r in b : <EOL> s += r * d . get ( u , <NUM_LIT:0> ) <EOL> return s <EOL> def cos ( ( l1 , l2 ) ) : <EOL> l1 = list ( l1 ) <EOL> l2 = list ( l2 ) <EOL> d2 = dict ( l2 ) <EOL> u2 = reverse ( l2 ) <EOL> for sid1 , us1 in l1 : <EOL> s = { } <EOL> for u , r in us1 : <EOL> if u in u2 : <EOL> for sid2 in u2 [ u ] : <EOL> if sid2 not in s : <EOL> s [ sid2 ] = ( vsum ( us1 , d2 [ sid2 ] ) , sid2 ) <EOL> sim = sorted ( s . values ( ) , reverse = True ) [ : <NUM_LIT:5> ] <EOL> yield sid1 , sim <EOL> final = rating . glom ( ) . cartesian ( rating . glom ( ) ) <EOL> print final . count ( ) <EOL> final = final . flatMap ( cos ) <EOL> final = final . reduceByKey ( lambda x , y : x + y ) . mapValue ( lambda x : sorted ( x , reverse = True ) [ : <NUM_LIT:5> ] ) <EOL> print '<STR_LIT>' , final . count ( ) </s>
