In [1]:

    

from pyspark import SparkConf, SparkContext

conf = SparkConf().setMaster("local").setAppName("PopularHero")
sc = SparkContext(conf = conf)


    

In [2]:

    

def countCoOccurences(line):
    elements = line.split()
    return (int(elements[0]), len(elements) - 1)

def parseNames(line):
    fields = line.split('\"')
    return (int(fields[0]), fields[1].encode("utf8"))


    

In [13]:

    

lines = sc.textFile("file:///opt/ipython/spark_code/example3/marvel-graph.txt")
lines.take(5)


    

    
Out[13]:





['5988 748 1722 3752 4655 5743 1872 3413 5527 6368 6085 4319 4728 1636 2397 3364 4001 1614 1819 1585 732 2660 3952 2507 3891 2070 2239 2602 612 1352 5447 4548 1596 5488 1605 5517 11 479 2554 2043 17 865 4292 6312 473 534 1479 6375 4456 ',
 '5989 4080 4264 4446 3779 2430 2297 6169 3530 3272 4282 6432 2548 4140 185 105 3878 2429 1334 4595 2767 3956 3877 4776 4946 3407 128 269 5775 5121 481 5516 4758 4053 1044 1602 3889 1535 6038 533 3986 ',
 '5982 217 595 1194 3308 2940 1815 794 1503 5197 859 5096 6039 2664 651 2244 528 284 1449 1097 1172 1092 108 3405 5204 387 4607 4545 3705 4930 1805 4712 4404 247 4754 4427 1845 536 5795 5978 533 3984 6056 ',
 '5983 1165 3836 4361 1282 716 4289 4646 6300 5084 2397 4454 1913 5861 5485 ',
 '5980 2731 3712 1587 6084 2472 2546 6313 875 859 323 2664 1469 522 2506 2919 2423 3624 5736 5046 1787 5776 3245 3840 2399 ']

In [14]:

    

names = sc.textFile("file:///opt/ipython/spark_code/example3/marvel-names.txt")
namesRdd = names.map(parseNames)
namesRdd.take(5)


    

    
Out[14]:





[(1, b'24-HOUR MAN/EMMANUEL'),
 (2, b'3-D MAN/CHARLES CHAN'),
 (3, b'4-D MAN/MERCURIO'),
 (4, b'8-BALL/'),
 (5, b'A')]

In [17]:

    

pairings = lines.map(countCoOccurences)
pairings.take(10)


    

    
Out[17]:





[(5988, 48),
 (5989, 40),
 (5982, 42),
 (5983, 14),
 (5980, 24),
 (5981, 17),
 (5986, 142),
 (5987, 81),
 (5984, 41),
 (5985, 19)]

In [18]:

    

totalFriendsByCharacter = pairings.reduceByKey(lambda x, y : x + y)
totalFriendsByCharacter.take(10)


    

    
Out[18]:





[(5988, 48),
 (5989, 40),
 (5982, 42),
 (5983, 14),
 (5980, 24),
 (5981, 17),
 (5986, 142),
 (5987, 81),
 (5984, 41),
 (5985, 19)]

In [19]:

    

flipped = totalFriendsByCharacter.map(lambda xy : (xy[1], xy[0]))
flipped.take(10)


    

    
Out[19]:





[(48, 5988),
 (40, 5989),
 (42, 5982),
 (14, 5983),
 (24, 5980),
 (17, 5981),
 (142, 5986),
 (81, 5987),
 (41, 5984),
 (19, 5985)]

In [10]:

    

mostPopular = flipped.max()

mostPopularName = namesRdd.lookup(mostPopular[1])[0]


    

In [22]:

    

print( "(id,name) ( {},{} ) is the most popular superhero".format(mostPopular[0],mostPopularName))


    

    




(id,name) ( 1933,b'CAPTAIN AMERICA' ) is the most popular superhero

In [ ]: