%cd ~/Documents/W261/hw9/

"""
Explanation: DATASCI W261: Machine Learning at Scale
Week 10, Homework 9
Group C: Katrina Adams, Eric Freeman, Doug Kelley
kradams@ischool.berkeley.edu, ericfreeman@ischool.berkeley.edu, kelleydac@ischool.berkeley.edu
3 November 2015
End of explanation
"""


%%writefile MRJob_NumberOfNodes.py
#!/usr/bin/env python
from mrjob.job import MRJob
from mrjob.step import MRStep   
from ast import literal_eval
from sys import maxint

class MRJobNumNodes(MRJob):
    

    def steps(self):
        mapper_only = 0
        if mapper_only:
            return [MRStep(mapper = self.mapper)]
        return [MRStep(
                mapper = self.mapper,
                reducer = self.reducer
                ),
               MRStep(
                reducer = self.reducer_count
                )]
    
    def mapper(self, _, line):
        line = line.strip().split('\t')
        key = line[0]
        value = literal_eval(line[1])
        destnodes = value.keys()
        
        yield key, None
        
        for node in destnodes:
            yield node, None
           
    def reducer(self,node,_):
        yield None, node
        
    def reducer_count(self,_,nodes):
        count = 0
        for node in nodes:
            count+=1
        yield None, count

        
if __name__ == '__main__':
    MRJobNumNodes.run()

%%writefile MRJob_PreprocessGraph.py
#!/usr/bin/env python
from mrjob.job import MRJob
from mrjob.step import MRStep   
from ast import literal_eval
from sys import maxint


class MRJobPreprocessGraph(MRJob):
    

    def steps(self):
        mapper_only = 0
        if mapper_only:
            return [MRStep(mapper_init = self.mapper_init,
                           mapper = self.mapper)]
        return [MRStep(
                mapper_init = self.mapper_init,
                mapper = self.mapper,
                reducer = self.reducer
                )]
    
    def mapper_init(self):
        N = 11
        self.PR = 1.0/N
        
    
    def mapper(self, _, line):
        line = line.strip().split('\t')
        key = line[0]
        edges = literal_eval(line[1])
        value = [self.PR, edges]
        yield key, value
        
        for node in edges.keys():
            yield node, [self.PR]

           
    def reducer(self,key,value):
        PR = 0
        edges = {}
        for v in value:
            if len(v)==1:
                PR = v[0]
            else:
                PR = v[0]
                edges = v[1]
        yield key, [PR, edges]

    
    '''
    def mapper(self, _, line):
        N = 11
        line = line.strip().split('\t')
        key = line[0]
        edges = literal_eval(line[1])
        PR = 1.0/N
        value = [PR, edges]
        yield None,{key:value}

           
    def reducer(self,_,data):
        for line in data:
            key = line.keys()[0]
            value = line[key]
            yield key, value
    '''
        
if __name__ == '__main__':
    MRJobPreprocessGraph.run()

!python MRJob_PreprocessGraph.py PageRank-test.txt > PageRank-test_input2.txt

!cat PageRank-test_input2.txt

%%writefile MRJob_PageRank.py
#!/usr/bin/env python
from mrjob.job import MRJob
from mrjob.step import MRStep   
from ast import literal_eval
from mrjob.compat import jobconf_from_env
import sys

class MRJobPageRank(MRJob):
    

    def steps(self):
        mapper_only = 0
        if mapper_only:
            return [MRStep(mapper = self.mapper)]
        
        
        return [MRStep(
                mapper = self.mapper_firstiter)]+\
                [MRStep(
                mapper = self.mapper,
                combiner = self.combiner,
                reducer = self.reducer)]*10
    
    def mapper_firstiter(self, _, line):
        line = line.strip().split('\t')
        key = line[0]
        value = literal_eval(line[1])

        PR = value[0]
        edges = value[1]
        
        yield key, [None,edges]
        
        destnodes = edges.keys()
        outdeg = len(destnodes)
        
        if outdeg==0:
            yield '*m',[PR]
        
        for node in destnodes:
            yield node, [1.0*PR/outdeg]
            

    def mapper(self, key, value):

        sys.stderr.write("MAPPER INPUT: ({0}, {1})\n".format(key,value))
        #sys.stderr.write("MAPPER")
        
        PR = value[0]
        edges = value[1]
        
        yield key, [None,edges]
        
        destnodes = edges.keys()
        outdeg = len(destnodes)
        
        if outdeg==0:
            yield '*m',[PR]
        
        for node in destnodes:
            yield node, [1.0*PR/outdeg]
            
            
    def combiner(self,node,data):
        PR = 0
        edges = {}
        for value in data:
            if value[0]==None:
                yield node, value
            else:
                PR+=value[0]
        yield node, [PR]
    
    
    def reducer(self,node,data):
        
        #sys.stderr.write("REDUCER NODE: {0}\n".format(node))
        
        G = int(jobconf_from_env('G'))
        alpha = float(jobconf_from_env('alpha'))
        
        PR = 0
        edges = {}
        
        for value in data:
            #yield node, value
            if value[0]==None:
                edges = value[1]
                
            else:
                PR+=value[0]
        
        if node=='*m':
            self.m = PR
        
        #sys.stderr.write("REDUCER PR: {0}\n".format(PR))
        
        else:
            PR_adj = alpha*(1.0/G)+(1-alpha)*(1.0*self.m/G+PR) 
            #yield node, PR
            #yield node, PR_adj
            
            #line = node+'\t'+str([PR_adj, edges])
            #yield line, ''
            
            yield node, [PR_adj, edges]

   
if __name__ == '__main__':
    MRJobPageRank.run()

%load_ext autoreload
%autoreload 2

from MRJob_NumberOfNodes import MRJobNumNodes
from MRJob_PreprocessGraph import MRJobPreprocessGraph
from MRJob_PageRank import MRJobPageRank

def numNodes(graphfilename):
    #mr_job = MRJobNumNodes(args=[graphfilename, "--strict-protocols", "-r", "emr", "--num-ec2-instances", "2", "--ec2-task-instance-type", "m1.small", "--pool-emr-job-flows", "--max-hours-idle=1"])
    mr_job_numnodes = MRJobNumNodes(args=[graphfilename])

    with mr_job_numnodes.make_runner() as runner:
        runner.run()
        for line in runner.stream_output():
            null,count =  mr_job_numnodes.parse_output_line(line)
            print "There are "+str(count)+" nodes in the graph."
                    
    return None

def preprocessGraph(graphfilename):
    #mr_job = MRJobNumNodes(args=[graphfilename, "--strict-protocols", "-r", "emr", "--num-ec2-instances", "2", "--ec2-task-instance-type", "m1.small", "--pool-emr-job-flows", "--max-hours-idle=1"])
    mr_job_preprocess = MRJobPreprocessGraph(args=[graphfilename])
    outputfilename = graphfilename.split('.')[0]+'_input.txt'
    with mr_job_preprocess.make_runner() as runner:
        with open(outputfilename, 'w') as f:
            runner.run()
            for line in runner.stream_output():
                node,value =  mr_job_preprocess.parse_output_line(line)
                f.write(node+'\t'+str(value)+'\n')
                    
    return None


def pagerank(graphinputfilename):
    #mr_job = MRJobNumNodes(args=[graphfilename, "--strict-protocols", "-r", "emr", "--num-ec2-instances", "2", "--ec2-task-instance-type", "m1.small", "--pool-emr-job-flows", "--max-hours-idle=1"])
    mr_job_pr1 = MRJobPageRank(args=[graphinputfilename,'--jobconf','alpha=0.15','--jobconf','G=11'])
    danglingmass= 0
    with mr_job_pr1.make_runner() as runner:
        runner.run()
        with open(graphinputfilename, 'w+') as f:
            for line in runner.stream_output():
                node,value =  mr_job_pr1.parse_output_line(line)
                f.write(node+'\t'+str(value)+'\n')
                print line

    return None




def hw9_1():

    graphfilename = "PageRank-test.txt"
    #numNodes(graphfilename)
    
    preprocessGraph(graphfilename)

    graphinputfilename = graphfilename.split('.')[0]+'_input.txt'
    
    #pagerank_step1(graphinputfilename, alpha, G)
    
    #pagerank_step1(graphinputfilename, alpha, G)
    
    for i in range(1):
        print "Iteration",i
        pagerank(graphinputfilename)
    
    
    
    
    return None
        
  

hw9_1()


"""
Explanation: HW9.0:
What is PageRank and what is it used for in the context of web search?
What modifications have to be made to the webgraph in order to leverage the machinery of Markov Chains to compute the steady stade distibuton?
OPTIONAL: In topic-specific pagerank, how can we insure that the irreducible property is satified? (HINT: see HW9.4)  
<font color='red'>
HW 9.0 ANSWER from Doug's notebook
</font>  
HW7.5.1:
Can we utilize combiners in the HW 7 to perform the shortest path implementation?  
Does order inversion help with the HW 7 shortest path implementation?
<font color='red'>
HW 7.5.1 ANSWER from Doug's notebook
</font>  

HW 9.1: MRJob implementation of basic PageRank  
Write a basic MRJob implementation of the iterative PageRank algorithm
that takes sparse adjacency lists as input (as explored in HW 7).
Make sure that your implementation utilizes teleportation (1-damping/the number of nodes in the network), and further, distributes the mass of dangling nodes with each iteration so that the output of each iteration is correctly normalized (sums to 1).
[NOTE: The PageRank algorithm assumes that a random surfer (walker), starting from a random web page, chooses the next page to which it will move by clicking at random, with probability d, one of the hyperlinks in the current page. This probability is represented by a so-called ‘damping factor’ d, where d ∈ (0, 1). Otherwise, with probability (1 − d), the surfer jumps to any web page in the network. If a page is a dangling end, meaning it has no outgoing hyperlinks, the random surfer selects an arbitrary web page from a uniform distribution and “teleports” to that page]  
As you build your code, use the test data  
s3://ucb-mids-mls-networks/PageRank-test.txt  
with teleportation parameter set to 0.15 (1-d, where d, the damping factor is set to 0.85), and crosscheck your work with the true result, displayed in the first image in the Wikipedia article:  
https://en.wikipedia.org/wiki/PageRank  
and here for reference are the corresponding PageRank probabilities:  
A,0.033
B,0.384
C,0.343
D,0.039
E,0.081
F,0.039
G,0.016
H,0.016
I,0.016
J,0.016
K,0.016
End of explanation
"""


!python MRJob_NumberOfNodes.py all-pages-indexed-out.txt -r emr --num-ec2-instances 10 --ec2-task-instance-type m1.medium

%%writefile MRJob_PreprocessGraph.py
#!/usr/bin/env python
from mrjob.job import MRJob
from mrjob.step import MRStep   
from ast import literal_eval
from mrjob.compat import jobconf_from_env


class MRJobPreprocessGraph(MRJob):
    

    def steps(self):
        mapper_only = 0
        if mapper_only:
            return [MRStep(mapper_init = self.mapper_init,
                           mapper = self.mapper)]
        return [MRStep(
                mapper_init = self.mapper_init,
                mapper = self.mapper,
                reducer = self.reducer
                )]
    
    def mapper_init(self):
        G = int(jobconf_from_env('G'))
        self.PR = 1.0/G
        
    
    def mapper(self, _, line):
        line = line.strip().split('\t')
        key = line[0]
        edges = literal_eval(line[1])
        value = [self.PR, edges]
        yield int(key), value
        
        for node in edges.keys():
            yield int(node), [self.PR]

           
    def reducer(self,key,value):
        PR = 0
        edges = {}
        for v in value:
            if len(v)==1:
                PR = v[0]
            else:
                PR = v[0]
                edges = v[1]
        yield int(key), [PR, edges]

    
        
if __name__ == '__main__':
    MRJobPreprocessGraph.run()

"""
Explanation: HW 9.2: Exploring PageRank teleportation and network plots
In order to overcome  problems such as disconnected components, the damping factor (a typical value for d is 0.85) can be varied.
Using the graph in HW1, plot the test graph (using networkx, https://networkx.github.io/) for several values of the damping parameter alpha, so that each nodes radius is proportional to its PageRank score. In particular you should do this for the following damping factors:   [0,0.25,0.5,0.75, 0.85, 1].
Note your plots should look like the following:  
https://en.wikipedia.org/wiki/PageRank#/media/File:PageRanks-Example.svg
<font color='red'>
HW 9.2 from Doug's notebook
</font>  
HW 9.3: Applying PageRank to the Wikipedia hyperlinks network
Run your PageRank implementation on the Wikipedia dataset for 10 iterations, and display the top 100 ranked nodes (with alpha = 0.85).  
Run your PageRank implementation on the Wikipedia dataset for 50 iterations, and display the top 100 ranked nodes (with teleportation factor of 0.15).
Have the top 100 ranked pages changed? Comment on your findings. Plot both 100 curves.
End of explanation
"""


!python MRJob_PreprocessGraph.py PageRank-testIndexed.txt --jobconf 'G=11' > PageRank-testIndexed_input.txt

%%writefile MRJob_PageRank.py
#!/usr/bin/env python
from mrjob.job import MRJob
from mrjob.step import MRStep   
from ast import literal_eval
from mrjob.compat import jobconf_from_env


class MRJobPageRank(MRJob):

    
    
    def configure_options(self):
        super(MRJobPageRank, self).configure_options()
        
        self.add_passthrough_option(
                '--iterations',
                dest = 'iterations',
                default = 10,
                type='int',
                help='number of iterations to run')
    

    def steps(self):        
        
        mapper_only = 0
        if mapper_only:
            return [MRStep(mapper = self.mapper)]
        
        return [MRStep(
                mapper = self.mapper_firstiter,
                reducer = self.reducer
                ),
               MRStep(
                reducer = self.reducer_redistr)]+\
                [MRStep(
                mapper = self.mapper,
                reducer = self.reducer
                ),
               MRStep(
                reducer = self.reducer_redistr)]*self.options.iterations
        
    
    def mapper_firstiter(self, _, line):
        line = line.strip().split('\t')
        key = line[0]
        value = literal_eval(line[1])

        PR = value[0]
        edges = value[1]
        
        yield int(key), [None,edges]
        
        destnodes = edges.keys()
        outdeg = len(destnodes)
        
        if outdeg==0:
            yield '*m',[PR]
        
        for node in destnodes:
            yield int(node), [1.0*PR/outdeg]
            

    def mapper(self, key, value):

        PR = value[0]
        edges = value[1]
        
        yield int(key), [None,edges]
        
        destnodes = edges.keys()
        outdeg = len(destnodes)
        
        if outdeg==0:
            yield '*m',[PR]
        
        for node in destnodes:
            yield int(node), [1.0*PR/outdeg]
  
            
    def reducer(self,node,data):
        
        alpha = float(jobconf_from_env('alpha'))
        G = int(jobconf_from_env('G'))
        
        PR = 0
        edges = {}
        for value in data:
            if len(value)==1:
                PR+=value[0]
            else:
                edges = value[1]
        
        if node=='*m':
            PR_adj = (1.0/G)*(1.0-alpha)*PR
            for i in range(G):
                yield i+1, [PR_adj]
        else:
            PR_adj = alpha*(1.0/G)+(1.0-alpha)*PR
            yield int(node), [PR_adj, edges]
        
    
    def reducer_redistr(self, node, data):
        
        PR = 0
        edges = {}
        for value in data:
            PR+=value[0]
            if len(value)>1:
                edges = value[1]
        
        yield int(node), [PR, edges]
        
        
   
if __name__ == '__main__':
    MRJobPageRank.run()

"""
Explanation: Preprocess Indexed Graph
End of explanation
"""


!python MRJob_PageRank.py PageRank-testIndexed_input.txt --jobconf alpha=0.15 --jobconf G=11 --iterations 50  

"""
Explanation: Test locally on small test graph
End of explanation
"""


!python MRJob_PageRank.py s3://ucb-mids-mls-katieadams/PageRank-testIndexed_input.txt --jobconf alpha=0.15 --jobconf G=11 --iterations 5 -r emr --num-ec2-instances 2 --ec2-task-instance-type m1.medium --output-dir s3://ucb-mids-mls-katieadams/outputtest --no-output  

"""
Explanation: Test on EMR with small test graph
End of explanation
"""


!python MRJob_PreprocessGraph.py s3://ucb-mids-mls-networks/wikipedia/all-pages-indexed-out.txt --jobconf G=15192277 -r emr --num-ec2-instances 15 --ec2-task-instance-type m1.medium --output-dir s3://ucb-mids-mls-katieadams/output-wiki-preprocess --no-output

"""
Explanation: Preprocess wikipedia graph
End of explanation
"""


!python MRJob_PageRank.py s3://ucb-mids-mls-katieadams/output-wiki-preprocess/* --jobconf alpha=0.85 --jobconf G=15192277 --iterations 9 -r emr --num-ec2-instances 15 --ec2-instance-type m1.medium --output-dir s3://ucb-mids-mls-katieadams/output-wiki-10x-2 --no-output  

"""
Explanation: PageRank on Wikipedia graph with alpha=0.85, 10 iterations
End of explanation
"""


!python MRJob_PageRank.py s3://ucb-mids-mls-katieadams/output-wiki-preprocess/* --jobconf alpha=0.15 --jobconf G=15192277 --iterations 49 -r emr --num-ec2-instances 15 --ec2-instance-type m1.medium --output-dir s3://ucb-mids-mls-katieadams/output-wiki-50x --no-output  

"""
Explanation: PageRank on Wikipedia graph with alpha=0.15, 50 iterations
Still running, 80% done, should be done my morning (barring any disasters)
End of explanation
"""


%%writefile MRJob_PreprocessGraph_topic.py
#!/usr/bin/env python
from mrjob.job import MRJob
from mrjob.step import MRStep   
from ast import literal_eval
from mrjob.compat import jobconf_from_env


class MRJobPreprocessGraphTopic(MRJob):
    

    def steps(self):
        mapper_only = 0
        if mapper_only:
            return [MRStep(mapper_init = self.mapper_init,
                           mapper = self.mapper)]
        return [MRStep(
                mapper_init = self.mapper_init,
                mapper = self.mapper,
                reducer = self.reducer
                )]
    
    def mapper_init(self):
        self.topic_labels = {}
        topicIdxFilename = jobconf_from_env('topicIdxFilename')
        with open(topicIdxFilename, 'r') as f:
            for line in f.readlines():
                line = line.strip()
                p,t = line.split('\t')
                self.topic_labels[int(p)]=int(t)
                
        G = int(jobconf_from_env('G'))
        self.PR_init = 1.0/G
        
    
    def mapper(self, _, line):
        line = line.strip().split('\t')
        key = int(line[0])
        edges = literal_eval(line[1])
        value = {}
        value['PR']=[self.PR_init]*11
        value['edges']=edges.keys()
        value['topic']=self.topic_labels[key]
            
        yield key, value
        
        for node in edges.keys():
            key=int(node)
            value = {}
            value['PR']=[self.PR_init]*11
            value['edges']=None
            value['topic']=self.topic_labels[key]
            yield key, value

           
    def reducer(self,key,value):
        data = {}
        data['topic']=None
        data['PR']=None
        data['edges']=[]
        for v in value:
            if data['topic']==None:
                data['topic']=v['topic']
            if v['edges']==None:
                data['PR']=v['PR']
            else:
                data['PR']=v['PR']
                data['edges']=v['edges']
        yield int(key), data

    
        
if __name__ == '__main__':
    MRJobPreprocessGraphTopic.run()

"""
Explanation: HW 9.4: Topic-specific PageRank implementation using MRJob
Modify your PageRank implementation to produce a topic specific PageRank implementation, as described in:
http://www-cs-students.stanford.edu/~taherh/papers/topic-sensitive-pagerank.pdf
Note in this article that there is a special caveat to ensure that the transition matrix is irreducible.
This caveat lies in footnote 3 on page 3:  
A minor caveat: to ensure that M is irreducible when p
contains any 0 entries, nodes not reachable from nonzero
nodes in p should be removed. In practice this is not problematic.

and must be adhered to for convergence to be guaranteed.  
Run topic specific PageRank on the following randomly generated network of 100 nodes:
s3://ucb-mids-mls-networks/randNet.txt  
which are organized into ten topics, as described in the file:
s3://ucb-mids-mls-networks/randNet_topics.txt  
Since there are 10 topics, your result should be 11 PageRank vectors (one for the vanilla PageRank implementation in 9.1, and one for each topic with the topic specific implementation). Print out the top ten ranking nodes and their topics for each of the 11 versions, and comment on your result.
Assume a teleportation factor of 0.15 in all your analyses.
End of explanation
"""


!python MRJob_PreprocessGraph_topic.py randNet.txt --file 'randNet_topics.txt' --jobconf 'G=100' --jobconf 'topicIdxFilename=randNet_topics.txt'> randNet_input.txt

"""
Explanation: Preprocess RandNet locally to include initialized page ranks and topics
End of explanation
"""


!python MRJob_PreprocessGraph_topic.py randNet.txt --file 'randNet_topics.txt' --jobconf 'G=100' --jobconf 'topicIdxFilename=randNet_topics.txt' -r emr --num-ec2-instances 2 --ec2-task-instance-type m1.medium --output-dir s3://ucb-mids-mls-katieadams/output-wiki-preprocess-topics --no-output

!head -1 randNet_input.txt

from collections import defaultdict
def numNodesByTopic(topicsIdxFile, topicsCountFile):
    numNodesTopic = defaultdict(int)
    with open('randNet_topics.txt', 'r') as f:
        for line in f.readlines():
            line = line.strip()
            p,t = line.split('\t')
            numNodesTopic[int(t)]+=1
            numNodesTopic['*']+=1
    with open('randNet_topicCount.txt','w') as f:
        for k,v in numNodesTopic.iteritems():
            line = str(k)+'\t'+str(v)+'\n'
            f.write(line)

            
topicsIdxFile, topicsCountFile
numNodesByTopic(topicsIdxFile, topicsCountFile)

!cat randNet_topicCount.txt

%%writefile MRJob_PageRank_Topic.py
#!/usr/bin/env python
from mrjob.job import MRJob
from mrjob.step import MRStep   
from ast import literal_eval
from mrjob.compat import jobconf_from_env
import sys


class MRJobPageRankTopic(MRJob):
    
    
    def configure_options(self):
        super(MRJobPageRankTopic, self).configure_options()
        
        self.add_passthrough_option(
                '--iterations',
                dest = 'iterations',
                default = 10,
                type='int',
                help='number of iterations to run')
    

    def steps(self):        
        
        mapper_only = 0
        if mapper_only:
            return [MRStep(mapper = self.mapper_firstiter)]
        
        return [MRStep(
                mapper = self.mapper_firstiter,
                reducer_init = self.reducer_init,
                reducer = self.reducer
                ),
               MRStep(
                reducer = self.reducer_redistr)]+\
                [MRStep(
                mapper = self.mapper,
                reducer_init = self.reducer_init,
                reducer = self.reducer
                ),
               MRStep(
                reducer = self.reducer_redistr)]*self.options.iterations
        
    
    def mapper_firstiter(self, _, line):
        
        line = line.strip().split('\t')
        key = line[0]
        value = literal_eval(line[1])
        
        graphvalue = {}
        graphvalue['PR']=None
        graphvalue['edges']=value['edges']
        graphvalue['topic']=value['topic']
        
        yield int(key), graphvalue
        
        destnodes = value['edges']
        outdeg = len(destnodes)
        
        sendvalue = {}
        sendvalue['PR']=[]
        sendvalue['topic']=None
        sendvalue['edges']=None
        
        if outdeg==0:
            sendvalue['PR']=value['PR']
            yield '*m',sendvalue
        else:
            
            for i in range(len(value['PR'])):
                PRi = value['PR'][i]
                sendvalue['PR'].append(1.0*PRi/outdeg)
            for node in destnodes:
                yield int(node), sendvalue
            

    def mapper(self, key, value):
       
        graphvalue = {}
        graphvalue['PR']=None
        graphvalue['edges']=value['edges']
        graphvalue['topic']=value['topic']
        
        yield int(key), graphvalue
        
        destnodes = value['edges']
        outdeg = len(destnodes)
        
        sendvalue = {}
        sendvalue['PR']=[]
        sendvalue['topic']=None
        sendvalue['edges']=None
        
        if outdeg==0:
            sendvalue['PR']=value['PR']
            yield '*m',sendvalue
        else:
            for i in range(len(value['PR'])):
                PRi = value['PR'][i]
                sendvalue['PR'].append(1.0*PRi/outdeg)
            for node in destnodes:
                yield int(node), sendvalue
  
    def reducer_init(self):
        
        self.alpha = float(jobconf_from_env('alpha'))
        self.G = int(jobconf_from_env('G'))
        self.n = int(jobconf_from_env('num_topics'))
        self.beta = float(jobconf_from_env('beta'))
        topicCountFilename = jobconf_from_env('topicCountFilename')
        
        self.T = {}
        with open(topicCountFilename, 'r') as f:
            for line in f.readlines():
                topic,count = line.strip().split('\t')
                if topic=='*':
                    topic = 0
                else: 
                    topic = int(topic)
                self.T[topic]=int(count)
            
    def reducer(self,node,data):
        
        sendvalue = {}
        sendvalue['PR']=[0]*(self.n+1)
        sendvalue['topic']=None
        sendvalue['edges']=None
       
        for value in data:
            if value['edges']==None:
                for i in range(len(value['PR'])):
                    sendvalue['PR'][i]+=value['PR'][i]
            else:
                sendvalue['topic']=int(value['topic'])
                sendvalue['edges']=value['edges']
        
        if node=='*m':
            PR_adj = []
            for i in range(self.n+1):
                PR_adj.append([(1.0/self.G)*(1.0-self.alpha)*sendvalue['PR'][i]])
            sendvalue['PR']=PR_adj
            for i in range(self.G):
                yield i+1, sendvalue
        else:
            PR_adj = []
            PR_adj.append(self.alpha*(1.0/self.G)+(1.0-self.alpha)*sendvalue['PR'][i])
            for i in range(1,self.n+1):
                if sendvalue['topic']==i:
                    PR_adj.append(self.alpha*self.beta*(1.0/self.T[i])+(1.0-self.alpha)*sendvalue['PR'][i])
                else:
                    PR_adj.append(self.alpha*(1-self.beta)*(1.0/(self.G-self.T[i]))+(1.0-self.alpha)*sendvalue['PR'][i])
                    
            sendvalue['PR']=PR_adj
            yield int(node), sendvalue
        
    
    def reducer_redistr(self, node, data):
        
        n = int(jobconf_from_env('num_topics'))
        
        finalvalue = {}
        finalvalue['PR']=[0]*(n+1)
        finalvalue['topic']=None
        finalvalue['edges']=None
        
        for value in data:
            for i in range(n+1):
                finalvalue['PR'][i]+=value['PR'][i]
            if value['edges']!=None:
                finalvalue['edges'] = value['edges']
                finalvalue['topic']=int(value['topic'])
        
        yield int(node), finalvalue
        
   
if __name__ == '__main__':
    MRJobPageRankTopic.run()

!python MRJob_PageRank_Topic.py randNet_input.txt --file 'randNet_topicCount.txt' --jobconf 'G=100' --jobconf 'alpha=0.15'\
                                                    --jobconf 'beta=0.99' --jobconf 'topicCountFilename=randNet_topicCount.txt'\
                                                    --jobconf 'num_topics=10' --iterations 9 > randNet_Topic_output.txt

%%writefile MRtop10.py
#!/usr/bin/env python
from mrjob.job import MRJob
from mrjob.protocol import JSONProtocol
from mrjob.step import MRStep
import re
from itertools import combinations
from ast import literal_eval
import sys
import string
from operator import itemgetter
from collections import defaultdict
from mrjob.compat import jobconf_from_env

'''
    Modified from Doug and Group E's code
'''


class MRtop10(MRJob):
    OUTPUT_PROTOCOL = JSONProtocol
    mapperListDict = defaultdict(list)
    reducerListDict = defaultdict(list)
    
    def jobconf(self):
        orig_jobconf = super(MRtop10, self).jobconf()        
        custom_jobconf = {
            'mapred.output.key.comparator.class': 'org.apache.hadoop.mapred.lib.KeyFieldBasedComparator',
            'mapred.text.key.comparator.options': '-k1rn',
            'mapred.reduce.tasks': '1',
        }
        combined_jobconf = orig_jobconf
        combined_jobconf.update(custom_jobconf)
        self.jobconf = combined_jobconf
        return combined_jobconf

    def steps(self):
        return [MRStep(
                mapper = self.mapper, 
                reducer_init = self.reducer_init,
                reducer = self.reducer,
                reducer_final = self.reducer_final
            )]
    def getFloatKey(item):
        return float(item[1])
    
    def mapper(self, key, line):

        fields = line.strip().split('\t')
        nodeid = fields[0]
        nodeDict = literal_eval(fields[1])
        for i in range(11):
            yield float(nodeDict['PR'][i]), (i, nodeid, nodeDict['topic'])

    def reducer_init(self):
        self.n = int(jobconf_from_env('num_topics'))
        self.top_pages = defaultdict(list)
            
    def reducer(self, score, nodetups):
        for node in nodetups:
            topic = node[0]
            if len(self.top_pages[topic])<10:
                self.top_pages[topic].append([score, node])

                
    def reducer_final(self):
        for page, scores in self.top_pages.iteritems():
            for score in scores:
                rank = score[0]
                node = score[1][1]
                topic = score[1][2]
                yield "Vector "+str(page), (rank, node, topic)

                
                
if __name__ == '__main__':
    MRtop10.run()

!python MRtop10.py randNet_Topic_output.txt --jobconf 'num_topics=10' -r emr

"""
Explanation: Test preprocessing on emr
End of explanation
"""


def assignWikiTopics():
    ftopics = open('wiki_topics.txt','w')
    i=0
    with open('indices.txt','r') as fidx:
        for line in fidx.readlines():
            i+=1
            if i%500000==0:
                print "progress: ",i*1.0/15192277
            line = line.strip().split('\t')
            topic = len(line[0])%10
            idx = int(line[1])
            ftopics.write(str(idx)+'\t'+str(topic)+'\n')
    ftopics.close()

assignWikiTopics()
            

!python MRJob_PreprocessGraph_topic.py s3://ucb-mids-mls-networks/wikipedia/all-pages-indexed-out.txt --file 'wiki_topics.txt' --jobconf G=15192277 --jobconf 'topicIdxFilename=wiki_topics.txt' -r emr --num-ec2-instances 4 --ec2-instance-type m1.large --output-dir s3://ucb-mids-mls-katieadams/output-wiki-preprocess-topics --no-output

"""
Explanation: HW 9.5: Applying topic-specific PageRank to Wikipedia
Here you will apply your topic-specific PageRank implementation to Wikipedia, defining topics (very arbitrarily) for each page by the length (number of characters) of the name of the article mod 10, so that there are 10 topics. Once again, print out the top ten ranking nodes and their topics for each of the 11 versions, and comment on your result.
Assume a teleportation factor of 0.15 in all your analyses.
End of explanation
"""


!python MRJob_PreprocessGraph_topic.py all-pages-indexed-out.txt --file 'wiki_topics.txt' --jobconf 'G=15192277' --jobconf 'topicIdxFilename=wiki_topics.txt' 

"""
Explanation: Memory Error on emr => running locally -- ran out of disk space in map step
End of explanation
"""


