پیاده سازی نگاشت کاهش (MapReduce) در پایتون – به زبان ساده

1import settings
2class MapReduce(object):
3    """MapReduce class representing the mapreduce model
4    note: the 'mapper' and 'reducer' methods must be
5    implemented to use the mapreduce model.
6    """
7    def __init__(self, input_dir = settings.default_input_dir, output_dir = settings.default_output_dir,
8                 n_mappers = settings.default_n_mappers, n_reducers = settings.default_n_reducers,
9                 clean = True):
10        """
11        :param input_dir: directory of the input files,
12        taken from the default settings if not provided
13        :param output_dir: directory of the output files,
14        taken from the default settings if not provided
15        :param n_mappers: number of mapper threads to use,
16        taken from the default settings if not provided
17        :param n_reducers: number of reducer threads to use,
18        taken from the default settings if not provided
19        :param clean: optional, if True temporary files are
20        deleted, True by default.
21        """
22        self.input_dir = input_dir
23        self.output_dir = output_dir
24        self.n_mappers = n_mappers
25        self.n_reducers = n_reducers
26        self.clean = clean
27    def mapper(self, key, value):
28        """outputs a list of key-value pairs, where the key is
29        potentially new and the values are of a potentially different type.
30        Note: this function is to be implemented.
31        :param key:
32        :param value:
33        """
34        pass
35    def reducer(self, key, values_list):
36        """Outputs a single value together with the provided key.
37        Note: this function is to be implemented.
38        :param key:
39        :param value_list:
40        """
41        pass

1def run_mapper(self, index):
2    """Runs the implemented mapper
3    :param index: the index of the thread to run on
4    """
5    # read a key
6    # read a value
7    # get the result of the mapper
8    # store the result to be used by the reducer
9    pass
10def run_reducer(self, index):
11    """Runs the implemented reducer
12    :param index: the index of the thread to run on
13    """
14    # load the results of the map
15    # for each key reduce the values
16    # store the results for this reducer 
17    pass
18def run(self):
19    """Executes the map and reduce operations
20    """
21    # initialize mappers list
22    map_workers = []
23    # initialize reducers list
24    rdc_workers = []
25    # run the map step
26    for thread_id in range(self.n_mappers):
27        p = Process(target=self.run_mapper, args=(thread_id,))
28        p.start()
29        map_workers.append(p)
30    [t.join() for t in map_workers]
31    # run the reduce step
32    for thread_id in range(self.n_reducers):
33        p = Process(target=self.run_reducer, args=(thread_id,))
34        p.start()
35        map_workers.append(p)
36    [t.join() for t in rdc_workers]

1class FileHandler(object):
2    """FileHandler class
3    Manages splitting input files and joining outputs together.
4    """
5    def __init__(self, input_file_path, output_dir):
6        """
7        Note: the input file path should be given for splitting.
8        The output directory is needed for joining the outputs.
9        :param input_file_path: input file path
10        :param output_dir: output directory path
11        """
12        self.input_file_path = input_file_path
13        self.output_dir = output_dir
14    def split_file(self, number_of_splits):
15        """split a file into multiple files.
16        :param number_of_splits: the number of splits.
17        """
18        pass
19    def join_files(self, number_of_files, clean = None, sort = True, decreasing = True):
20        """join all the files in the output directory into a
21        single output file.
22        :param number_of_files: total number of files.
23        :param clean: if True the reduce outputs will be deleted,
24        by default takes the value of self.clean.
25        :param sort: sort the outputs.
26        :param decreasing: sort by decreasing order, high value
27        to low value.
28        :return output_join_list: a list of the outputs
29        """
30        pass

1import os
2import json
3
4class FileHandler(object):
5    """FileHandler class
6    Manages splitting input files and joining outputs together.
7    """
8    def __init__(self, input_file_path, output_dir):
9        """
10        Note: the input file path should be given for splitting.
11        The output directory is needed for joining the outputs.
12        :param input_file_path: input file path
13        :param output_dir: output directory path
14        """
15        self.input_file_path = input_file_path
16        self.output_dir = output_dir
17    def begin_file_split(self, split_index, index):
18        """initialize a split file by opening and adding an index.
19        :param split_index: the split index we are currently on, to be used for naming the file.
20        :param index: the index given to the file.
21        """
22        file_split = open(settings.get_input_split_file(split_index-1), "w+")
23        file_split.write(str(index) + "\n")
24        return file_split
25    def is_on_split_position(self, character, index, split_size, current_split):
26        """Check if it is the right time to split.
27        i.e: character is a space and the limit has been reached.
28        :param character: the character we are currently on.
29        :param index: the index we are currently on.
30        :param split_size: the size of each single split.
31        :param current_split: the split we are currently on.
32        """
33        return index>split_size*current_split+1 and character.isspace()
34    def split_file(self, number_of_splits):
35        """split a file into multiple files.
36        note: this has not been optimized to avoid overhead.
37        :param number_of_splits: the number of chunks to
38        split the file into.
39        """
40        file_size = os.path.getsize(self.input_file_path)
41        unit_size = file_size / number_of_splits + 1
42        original_file = open(self.input_file_path, "r")
43        file_content = original_file.read()
44        original_file.close()
45        (index, current_split_index) = (1, 1)
46        current_split_unit = self.begin_file_split(current_split_index, index)
47        for character in file_content:
48            current_split_unit.write(character)
49            if self.is_on_split_position(character, index, unit_size, current_split_index):
50                current_split_unit.close()
51                current_split_index += 1
52                current_split_unit = self.begin_file_split(current_split_index,index)
53            index += 1
54        current_split_unit.close()

1def run_mapper(self, index):
2    """Runs the implemented mapper
3    :param index: the index of the thread to run on
4    """
5    input_split_file = open(settings.get_input_split_file(index), "r")
6    key = input_split_file.readline()
7    value = input_split_file.read()
8    input_split_file.close()
9    if(self.clean):
10        os.unlink(settings.get_input_split_file(index))
11    mapper_result = self.mapper(key, value)
12    for reducer_index in range(self.n_reducers):
13        temp_map_file = open(settings.get_temp_map_file(index, reducer_index), "w+")
14        json.dump([(key, value) for (key, value) in mapper_result 
15                                    if self.check_position(key, reducer_index)]
16                    , temp_map_file)
17        temp_map_file.close()
18    
19def run_reducer(self, index):
20    """Runs the implemented reducer
21    :param index: the index of the thread to run on
22    """
23    key_values_map = {}
24    for mapper_index in range(self.n_mappers):
25        temp_map_file = open(settings.get_temp_map_file(mapper_index, index), "r")
26        mapper_results = json.load(temp_map_file)
27        for (key, value) in mapper_results:
28            if not(key in key_values_map):
29                key_values_map[key] = []
30            try:
31                key_values_map[key].append(value)
32            except Exception, e:
33                print "Exception while inserting key: "+str(e)
34        temp_map_file.close()
35        if self.clean:
36            os.unlink(settings.get_temp_map_file(mapper_index, index))
37    key_value_list = []
38    for key in key_values_map:
39        key_value_list.append(self.reducer(key, key_values_map[key]))
40    output_file = open(settings.get_output_file(index), "w+")
41    json.dump(key_value_list, output_file)
42    output_file.close()

1def run(self, join=False):
2    """Executes the map and reduce operations
3    :param join: True if we need to join the outputs, False by default.
4    """
5    # initialize mappers list
6    map_workers = []
7    # initialize reducers list
8    rdc_workers = []
9    # run the map step
10    for thread_id in range(self.n_mappers):
11        p = Process(target=self.run_mapper, args=(thread_id,))
12        p.start()
13        map_workers.append(p)
14    [t.join() for t in map_workers]
15    # run the reduce step
16    for thread_id in range(self.n_reducers):
17        p = Process(target=self.run_reducer, args=(thread_id,))
18        p.start()
19        map_workers.append(p)
20    [t.join() for t in rdc_workers]
21    if join:
22        self.join_outputs()

1import os
2import json
3import settings
4from multiprocessing import Process
5
6
7class FileHandler(object):
8    """FileHandler class
9    Manages splitting input files and joining outputs together.
10    
11    """
12    def __init__(self, input_file_path, output_dir):
13        """
14        Note: the input file path should be given for splitting.
15        The output directory is needed for joining the outputs.
16        :param input_file_path: input file path
17        :param output_dir: output directory path
18        """
19        self.input_file_path = input_file_path
20        self.output_dir = output_dir
21
22    def initiate_file_split(self, split_index, index):
23        """initialize a split file by opening and adding an index.
24        :param split_index: the split index we are currently on, to be used for naming the file.
25        :param index: the index given to the file.
26        """
27        file_split = open(settings.get_input_split_file(split_index-1), "w+")
28        file_split.write(str(index) + "\n")
29        return file_split
30
31    def is_on_split_position(self, character, index, split_size, current_split):
32        """Check if it is the right time to split.
33        i.e: character is a space and the limit has been reached.
34        :param character: the character we are currently on.
35        :param index: the index we are currently on.
36        :param split_size: the size of each single split.
37        :param current_split: the split we are currently on.
38        """
39        return index>split_size*current_split+1 and character.isspace()
40
41    def split_file(self, number_of_splits):
42        """split a file into multiple files.
43        note: this has not been optimized to avoid overhead.
44        :param number_of_splits: the number of chunks to
45        split the file into.
46        """
47        file_size = os.path.getsize(self.input_file_path)
48        unit_size = file_size / number_of_splits + 1
49        original_file = open(self.input_file_path, "r")
50        file_content = original_file.read()
51        original_file.close()
52        (index, current_split_index) = (1, 1)
53        current_split_unit = self.initiate_file_split(current_split_index, index)
54        for character in file_content:
55            current_split_unit.write(character)
56            if self.is_on_split_position(character, index, unit_size, current_split_index):
57                current_split_unit.close()
58                current_split_index += 1
59                current_split_unit = self.initiate_file_split(current_split_index, index)
60            index += 1
61        current_split_unit.close()
62        
63    def join_files(self, number_of_files, clean = False, sort = True, decreasing = True):
64        """join all the files in the output directory into a
65        single output file.
66        :param number_of_files: total number of files.
67        :param clean: if True the reduce outputs will be deleted,
68        by default takes the value of self.clean.
69        :param sort: sort the outputs.
70        :param decreasing: sort by decreasing order, high value
71        to low value.
72        :return output_join_list: a list of the outputs
73        """
74        output_join_list = []
75        for reducer_index in xrange(0, number_of_files):
76            f = open(settings.get_output_file(reducer_index), "r")
77            output_join_list += json.load(f)
78            f.close()
79            if clean:
80                os.unlink(settings.get_output_file(reducer_index))
81        if sort:
82            from operator import itemgetter as operator_ig
83            # sort using the key
84            output_join_list.sort(key=operator_ig(1), reverse=decreasing)
85        output_join_file = open(settings.get_output_join_file(self.output_dir), "w+")
86        json.dump(output_join_list, output_join_file)
87        output_join_file.close()
88        return output_join_list
89    
90class MapReduce(object):
91    """MapReduce class
92    Note: mapper and reducer functions need to be implemented.
93    
94    """
95
96    def __init__(self, input_dir = settings.default_input_dir, output_dir = settings.default_output_dir,
97                 n_mappers = settings.default_n_mappers, n_reducers = settings.default_n_reducers,
98                 clean = True):
99        """
100        :param input_dir: directory of the input files,
101        taken from the default settings if not provided
102        :param output_dir: directory of the output files,
103        taken from the default settings if not provided
104        :param n_mappers: number of mapper threads to use,
105        taken from the default settings if not provided
106        :param n_reducers: number of reducer threads to use,
107        taken from the default settings if not provided
108        :param clean: optional, if True temporary files are
109        deleted, True by default.
110        """
111        self.input_dir = input_dir
112        self.output_dir = output_dir
113        self.n_mappers = n_mappers
114        self.n_reducers = n_reducers
115        self.clean = clean
116        self.file_handler = FileHandler(settings.get_input_file(self.input_dir), self.output_dir)
117        self.file_handler.split_file(self.n_mappers)
118
119    def mapper(self, key, value):
120        """outputs a list of key-value pairs, where the key is
121        potentially new and the values are of a potentially different type.
122        Note: this function is to be implemented.
123        :param key: key of the current mapper
124        :param value: value for the corresponding key
125        note: this method should be implemented
126        """
127        pass
128
129    def reducer(self, key, values_list):
130        """Outputs a single value together with the provided key.
131        Note: this function is to be implemented.
132        :param key: key of the reducer
133        :param value_list: list of values for the key
134        note: this method should be implemented
135        """
136        pass
137
138    def check_position(self, key, position):
139        """Checks if we are on the right position
140        """
141        return position == (hash(key) % self.n_reducers)
142
143    def run_mapper(self, index):
144        """Runs the implemented mapper
145        :param index: the index of the thread to run on
146        """
147        input_split_file = open(settings.get_input_split_file(index), "r")
148        key = input_split_file.readline()
149        value = input_split_file.read()
150        input_split_file.close()
151        if(self.clean):
152            os.unlink(settings.get_input_split_file(index))
153        mapper_result = self.mapper(key, value)
154        for reducer_index in range(self.n_reducers):
155            temp_map_file = open(settings.get_temp_map_file(index, reducer_index), "w+")
156            json.dump([(key, value) for (key, value) in mapper_result 
157                                        if self.check_position(key, reducer_index)]
158                        , temp_map_file)
159            temp_map_file.close()
160        
161    def run_reducer(self, index):
162        """Runs the implemented reducer
163        :param index: the index of the thread to run on
164        """
165        key_values_map = {}
166        for mapper_index in range(self.n_mappers):
167            temp_map_file = open(settings.get_temp_map_file(mapper_index, index), "r")
168            mapper_results = json.load(temp_map_file)
169            for (key, value) in mapper_results:
170                if not(key in key_values_map):
171                    key_values_map[key] = []
172                try:
173                    key_values_map[key].append(value)
174                except Exception, e:
175                    print "Exception while inserting key: "+str(e)
176            temp_map_file.close()
177            if self.clean:
178                os.unlink(settings.get_temp_map_file(mapper_index, index))
179        key_value_list = []
180        for key in key_values_map:
181            key_value_list.append(self.reducer(key, key_values_map[key]))
182        output_file = open(settings.get_output_file(index), "w+")
183        json.dump(key_value_list, output_file)
184        output_file.close()
185
186    def run(self, join=False):
187        """Executes the map and reduce operations
188        :param join: True if we need to join the outputs, False by default.
189        """
190        # initialize mappers list
191        map_workers = []
192        # initialize reducers list
193        rdc_workers = []
194        # run the map step
195        for thread_id in range(self.n_mappers):
196            p = Process(target=self.run_mapper, args=(thread_id,))
197            p.start()
198            map_workers.append(p)
199        [t.join() for t in map_workers]
200        # run the reduce step
201        for thread_id in range(self.n_reducers):
202            p = Process(target=self.run_reducer, args=(thread_id,))
203            p.start()
204            map_workers.append(p)
205        [t.join() for t in rdc_workers]
206        if join:
207            self.join_outputs()
208        
209    def join_outputs(self, clean = True, sort = True, decreasing = True):
210        """Join all the reduce output files into a single output file.
211        
212        :param clean: if True the reduce outputs will be deleted, by default takes the value of self.clean
213        :param sort: sort the outputs
214        :param decreasing: sort by decreasing order, high value to low value
215        
216        """
217        try:
218            return self.file_handler.join_files(self.n_reducers, clean, sort, decreasing)
219        except Exception, e:
220            print "Exception occured while joining: maybe the join has been performed already  -- "+str(e)
221            return []

1# set default directory for the input files
2default_input_dir = "input_files"
3# set default directory for the temporary map files
4default_map_dir = "temp_map_files"
5# set default directory for the output files
6default_output_dir = "output_files"
7# set default number for the map and reduce threads
8default_n_mappers = 4
9default_n_reducers = 4
10# return the name of the input file to be split into chunks
11def get_input_file(input_dir = None, extension = ".ext"):
12    if not(input_dir is None):
13        return input_dir+"/file" + extension
14    return default_input_dir + "/file" + extension
15    
16    
17# return the name of the current split file corresponding to the given index
18def get_input_split_file(index, input_dir = None, extension = ".ext"):
19    if not(input_dir is None):
20        return input_dir+"/file_"+ str(index) + extension
21    return default_input_dir + "/file_" + str(index) + extension
22        
23        
24# return the name of the temporary map file corresponding to the given index
25def get_temp_map_file(index, reducer, output_dir = None, extension = ".ext"):
26    if not(output_dir is None):
27        return output_dir + "/map_file_" + str(index)+"-" + str(reducer) + extension
28    return default_output_dir + "/map_file_" + str(index) + "-" + str(reducer) + extension
29
30# return the name of the output file given its corresponding index
31def get_output_file(index, output_dir = None, extension = ".out"):
32    if not(output_dir is None):
33        return output_dir+"/reduce_file_"+ str(index) + extension
34    return default_output_dir + "/reduce_file_" + str(index) + extension
35        
36# return the name of the output file
37def get_output_join_file(output_dir = None, extension = ".out"):
38    if not(output_dir is None):
39        return output_dir +"/output" + extension
40    return default_output_dir + "/output" + extension

1from mapreduce import MapReduce
2import sys
3class WordCount(MapReduce):
4    def __init__(self, input_dir, output_dir, n_mappers, n_reducers):
5        MapReduce.__init__(self,  input_dir, output_dir, n_mappers, n_reducers)
6    def mapper(self, key, value):
7        """Map function for the word count example
8        Note: Each line needs to be separated into words, and each word
9        needs to be converted to lower case.
10        """
11        results = []
12        default_count = 1
13        # seperate line into words
14        for word in value.split():
15            if self.is_valid_word(word):
16                # lowercase words