Answer to CSE Handson 2
-
Answer 1
The numbers of map workers and reduce workers.
-
Answer 2
- Init. Set maptask and reducetask parameters and split the input file into serverl small subfiles for parallel usage.
- Run. Create a process pool.
- Use pool.map function to call the WordCount's doMap function parallelly.
- The doMap function calls user's map function and creates intermediate files which store some key/value pairs.
- Then another pool.map calls the WordCount's doReduce function parallelly.
- The doReduce function calls user's reduce function. Reduce function uses those intermediate files as input and does reduce operation.
- Merge the result.
-
Answer 3
-
Key: the offset of a chunk of file contents
-
Value: a chunk of file contents.
-
-
Answer 4
-
Key: lower-case version of title-cased word
-
Value: pairs like {word, 1}, and the word is the same as the key.
-
-
Answer 5
They are 4 times and 2 times. The size of iterable tells the invocation numbers. In this case, maptask is 4 and reducetask is 2.
def run(self): pool = Pool(processes=max(self.maptask, self.reducetask),) regions = pool.map(self.doMap, range(0, self.maptask)) partitions = pool.map(self.doReduce, range(0, self.reducetask))map(func, iterable[, chunksize])
A parallel equivalent of the map() built-in function (it supports only one iterable argument though). It blocks until the result is ready.
This method chops the iterable into a number of chunks which it submits to the process pool as separate tasks. The (approximate) size of these chunks can be specified by setting chunksize to a positive integer.
-
Answer 6
The four invocations of doMap run in parallel. The two invocation of doReduce run in parallel. However, doMap and doReduce do not run in parallel because the map function blocks until the result is ready.
-
Answer 7
-
In theory, it's about 1,208,690 bytes, which can be calculated from:
size = os.stat(self.path).st_size; chunk = size / self.maptask chunk += 1 ## check the chunk size here and get the result -
In fact, the actual size can be recorded by:
def doMap(self, i): f = open("#split-%s-%s" % (self.path, i), "r") keyvalue = f.readline() value = f.read() f.close() print(len(value))So the size is (1,208,691+1,208,697+1,208,683+1,208,686) / 4 = 1 ,208,689.25
-
-
Answer 8
def doReduce(self, i): keys = {} out = [] count = 0; for m in range(0, self.maptask): # print "reduce", i, "#map-%s-%s-%d" % (self.path, m, i) f = open("#map-%s-%s-%d" % (self.path, m, i), "r") itemlist = pickle.load(f) for item in itemlist: if keys.has_key(item[0]): keys[item[0]].append(item) else: keys[item[0]] = [item] count = count + 1我认为题目问的是key的数目而不是item的数目，如果问的是iem的数目，把count = count +1 放到上面的if中就好。
I used a variable to record the keys processed and got 2,250 keys and 2,222 keys as a result of two doReduce. So the average is (2260 + 2222) / 2 = 2,236 keys per doReduce.
-
Answer 9
➜ handson-2 python mapreduce.py kjv12.txt maptask: 1 reducetask: 1 time: 2.28125s maptask: 1 reducetask: 2 time: 2.421875s maptask: 1 reducetask: 3 time: 2.328125s maptask: 1 reducetask: 4 time: 2.3125s maptask: 1 reducetask: 5 time: 2.28125s maptask: 1 reducetask: 6 time: 2.234375s maptask: 2 reducetask: 1 time: 2.25s maptask: 2 reducetask: 2 time: 2.234375s maptask: 2 reducetask: 3 time: 2.21875s maptask: 2 reducetask: 4 time: 2.265625s maptask: 2 reducetask: 5 time: 2.28125s maptask: 2 reducetask: 6 time: 2.328125s maptask: 3 reducetask: 1 time: 2.296875s maptask: 3 reducetask: 2 time: 2.25s maptask: 3 reducetask: 3 time: 2.46875s maptask: 3 reducetask: 4 time: 2.375s maptask: 3 reducetask: 5 time: 2.4375s maptask: 3 reducetask: 6 time: 2.3125s maptask: 4 reducetask: 1 time: 2.46875s maptask: 4 reducetask: 2 time: 2.4375s maptask: 4 reducetask: 3 time: 2.4375s maptask: 4 reducetask: 4 time: 2.296875s maptask: 4 reducetask: 5 time: 2.359375s maptask: 4 reducetask: 6 time: 2.34375s maptask: 5 reducetask: 1 time: 2.546875s maptask: 5 reducetask: 2 time: 2.546875s maptask: 5 reducetask: 3 time: 2.375s maptask: 5 reducetask: 4 time: 2.359375s maptask: 5 reducetask: 5 time: 2.40625s maptask: 5 reducetask: 6 time: 2.40625s maptask: 6 reducetask: 1 time: 2.53125s maptask: 6 reducetask: 2 time: 2.453125s maptask: 6 reducetask: 3 time: 2.4375s maptask: 6 reducetask: 4 time: 2.4375s maptask: 6 reducetask: 5 time: 2.421875s maptask: 6 reducetask: 6 time: 2.75s呈现先减后增的规律，我的电脑是双核的，每次两个task是比较适合电脑的并行性，所以当maptask/reducetask在2左右的时候，速度最快。
当(maptask/reducetask - CPU cores)越大时，会产生更大的overhead，此时速度反而变慢。
-
Answer 10
class ReverseIndex(MapReduce): def __init__(self, maptask, reducetask, path): MapReduce.__init__(self, maptask, reducetask, path) # Produce a (key, value) pair for each title word in value def Map(self, keyvalue, value): results = [] i = 0 n = len(value) while i < n: # skip non-ascii letters in C/C++ style a la MapReduce paper: while i < n and value[i] not in string.ascii_letters: i += 1 start = i while i < n and value[i] in string.ascii_letters: i += 1 w = value[start:i] if start < i and w.istitle(): results.append ((w.lower(), int(keyvalue) + start)) return results # Reduce [(key,value), ...]) def Reduce(self, key, keyvalues): return (key, [pair[1] for pair in keyvalues])
