popular_movies.py

'''
__author__ = 'anis016'
Date: 18.05.16
Time: 20:41
'''
import os
import sys
from mrjob.job import MRJob
from mrjob.step import MRStep


class MRPopularMovie(MRJob):

    def configure_options(self):
        super(MRPopularMovie, self).configure_options()
        self.add_file_option('--items', help='Path to u.item') # waits for --items followed by data

    def steps(self):
        return [
            MRStep(mapper=self.mapper_get_ratings,
                   reducer_init=self.reducer_init,
                   reducer=self.reducer_count_view),
            MRStep(reducer=self.reducer_most_viewed)
        ]

    def mapper_get_ratings(self, key, line):
        (user_id, movie_id, rating, timestamp) = line.split('\t')
        yield movie_id, 1

    def reducer_init(self):
        self.movie_name = {}
        with open('u.item', encoding="ISO-8859-1") as f:
            # print('type of file', type(f))
            for line in f:
                fields = line.split('|')
                self.movie_name[fields[0]] = fields[1]

    def reducer_count_view(self, movie_id, occurences):
        yield None, (sum(occurences),  self.movie_name[movie_id])

    def reducer_most_viewed(self, key, movie_view_group):
        yield max(movie_view_group)

if __name__ == '__main__':

    if len(sys.argv) == 1:
        #  add data file path and pass in the argument
        data_file = 'data/ml-100k/u.data'
        item_file = 'data/ml-100k/u.item'

        data_path = os.path.join(os.path.dirname(os.getcwd()), data_file)
        item_path = os.path.join(os.path.dirname(os.getcwd()), item_file)
        sys.argv.append(data_path)

        sys.argv.append('--items')
        sys.argv.append(item_path)

    MRPopularMovie.run()