MapReduce : Passing additional file

In an earlier post, we discussed passing additional parameters to MapReduce Job. But there are cases in which we will have to pass some additional files during MapReduce. But since MapReduce runs in multiple nodes, we need to ensure that this additional file that mapper/reducer refers is in that particular node in which its running.  In this post we will disucss how to handle this. Let us say we need to find most popular movie from movie-lens database. If you download movie-lens data, there are 2 files in which we are interested in .  ( and u.item). The format of file is as shown here..
userid  movieid rating  timestamp 
196	242	3	881250949
186	302	3	891717742
22	377	1	878887116
244	51	2	880606923
166	346	1	886397596

movieID  Name Release Date  URL
1|Toy Story (1995)|01-Jan-1995|||...
2|GoldenEye (1995)|01-Jan-1995|||..
3|Four Rooms (1995)|01-Jan-1995|||...
4|Get Shorty (1995)|01-Jan-1995|||..


from mrjob.job import MRJob
from mrjob.step import MRStep

    Goal is to calculate most popular movie from the movie lens data 
    Algorithm :  Most watched movie is most popular movie
    Also pass u.item file along with job, so that it will be available in    each nodes, so we can map movieID to a moviename

class MRMostPopularMovieName(MRJob):

    def configure_options(self):
        super(MRMostPopularMovieName, self).configure_options()
        self.add_file_option('--items', help='path to u.item') 

    def steps(self):
        return [
                     MRStep(mapper = self.mapper_get_count,
                            reducer_init = self.reducer_init,
                            reducer = self.reducer_sum_count),

                     MRStep( reducer = self.reducer_max_count)


    def mapper_get_count(self, _ , line):
        userId, movieID, rating, timeStamp = line.split('\t')
        yield movieID, 1

    def reducer_init(self):
        self.movie_names = {}

        with open("u.item") as f:
            for line in f:
                fields = line.split('|')
                self.movie_names[fields[0]] = fields[1]

    def reducer_sum_count(self, movieID, values):
        yield None , (sum(values), self.movie_names[movieID])

    def reducer_max_count(self, _ , views):
        yield max(views)

if __name__ == '__main__':
>> python --items data\ml-100k\u.item data\ml-100k\

Configure Options In the function configure_option  we have used add file option to specify that a file will be passed along with this MapReduce Job and should be copied to each node.  We have also mentioned that the file will be identified by –items in the command line arg along with the actual data file.

MRJob Step We have also specified some additional steps including a reducer_init which will get called before the reducer in that step gets called. In reducer_init, we create a dictionary of movie names from u.item file which will be present in the same node where mapper/reducer is running.

Another important thing to note in this example is that reducer is passing ‘None’ as key and a tuple of  sum(views) and movie_name  (sum(values), self.movie_names[movieID]) as value. This  will help us to calculate the max(views) in the next reducer, which will produce a single result.