目录
1. MapReduce概述
2. 极简MapReduce内存版
3. 复杂MapReduce磁盘版
4. MapReduce思想的总结
1. MapReduce概述
以前写过一篇 MapReduce思想 ，这次再深入一点，简单实现一把单机内存的。MapReduce就是把它理解成高阶函数，需要传入map和reduce所需的函数，即可对一个集合的键值对进行按需变换。
我们按Python的map和reduce逻辑，接收这两个高阶函数所需的函数形态，而不是像Java那样写一个Mapper/Reducer，当然如果需要复杂一点map功能，可以通过闭包来实现。
2. 极简MapReduce内存版
如果是效仿Python，很容易想到实现MapReduce 就是map + 分组 + reduce。一头一尾的map和reduce都可以直接用py的内置函数，中间分组也容易用groupby实现。因此首先约定map和reduce的自定义函数写法就是直接按内置函数要求来写，只不过map输出是键值对，要求用tuple/list 来区分key和value；reduce时候只对一组的values做自定义聚合，不能操作key。到这里很容易实现严格的MapReduce，你用起来大概就和Spark提供的mapreduce一样了。如果想直接做wordcount好像不太容易，因为一行文本得输出一个词表数组，因此可以宽松一点，实现类似flatMap的效果，也就是即使map只输出一个，也要放进列表里，由框架去展开。
def wc_map(line):
return [(x,1) for x in line.strip().split()]reduce就是严格的两个参数和一个返回值形式（其中y表示上一轮迭代的结果，x表示每次便利数组拿到的值，这种严格的reduce函数并容易设计）
def wc_reduce(y, x):
return y+x剩下要做的事就是把map出来的数据做 拉平和排序分组，最后对每一组做reduce。
from itertools import groupby
from functools import reduce
def map_reduce(f_map, f_reduce, seq):
map_result = map(f_map, seq)
flattened = sum(map_result, []) #拉平
shuffled = sorted(flattened, key= lambda x:x[0]) #排序
mr_result = [(key, reduce(f_reduce, (x[1] for x in data))) for key, data in groupby( shuffled, key=lambda x:x[0])] #groupby分组
return mr_result3. 复杂MapReduce磁盘版
如果内存不够，就必须不断把数据写到磁盘上，也就是过去大数据面试题最喜欢考察的点。那些题若是有单机版MapReduce，基本都可以直接做出来。大体结构如下图：
其中partition只有一个，就不用单独设计了。过程也简化很多，也就是每当buffer满了，就排序写出到成一个数据块，最后将所有的数据块merge起来。merge过程写起来不是很容易，需要维持所有数据块的句柄数组有序：每次读取所有句柄中最小的那一条，随后数组需要重新按最小值排序，直到所有数据块读取完。
另外就是因为中间数据需要落地，所以这一版的MapReduce框架直接从文件到文件。那么让写map和reduce函数直接对着文件输出，类似Java那样提供一个context的参数呢？我选择允许用yield来输出内容，这样相当于也宽松了输出多个键值对的要求。
def wc_map(line):
for t in line.strip().split():
yield (t, 1)最后同样宽松reduce的写法，不用写那么复杂的两参数一返回，而是像Java那样，拿到一组key和可迭代的values，甚至可以这一组key输出多个结果。
def reduce_func(key , kv_pairs):
s = 0
for k,v in kv_pairs :
s += v
yield (key, s)很显然我们要补充很多东西，要读取文件、要设计缓存、要做merge，因此设计了一个类，过程就不解释了
from operator import itemgetter
import itertools
import sys
class MapReduce:
class Buffer:
def __init__(self, buffer_size = 80000):
self.buffer_size = buffer_size
self.buffer = []
self.current_size = 0
def add_data(self, kv) -> bool :
self.current_size += len(str(kv))
self.buffer.append(kv)
return True if self.current_size > self.buffer_size else False
def spill(self, file_path):
if self.buffer:
self.buffer.sort(key = lambda x:x[0])
with open(file_path, mode='w') as f:
for kv in self.buffer:
f.write(str(kv) + '\n')
self.buffer = []
self.current_size = 0
def __init__(self, buffer_size,temp_dir):
self.buffer = self.Buffer(buffer_size)
self.temp_dir = temp_dir
self.block_id = 0
def __fetch_kvs_from_map(self, map_f, infile):
with open(infile, mode='r', encoding='utf-8') as f:
for line in f:
for k,v in map_f(line):
yield (k,v)
def __run_map(self, map_f, infile):
block_files = [x for x in os.listdir(self.temp_dir) if x.startswith('block.') and x.split('.',1)[1].isdigit() ]
for block in block_files:
os.remove(f'{self.temp_dir}/{block}')
for kv in self.__fetch_kvs_from_map(map_f, infile):
if self.buffer.add_data(kv):
self.buffer.spill(f'{self.temp_dir}/block.{self.block_id}')
self.block_id += 1
self.buffer.spill(f'{self.temp_dir}/block.{self.block_id}')
def __merge_sort_from_files(self):
block_files = [x for x in os.listdir(self.temp_dir) if x.startswith('block.') and x.split('.',1)[1].isdigit() ]
block_files.sort(key = lambda filename: int(filename.split('.',1)[1]))
ffs = [open(f'{self.temp_dir}/{block}', 'r') for block in block_files]
first_kvs = [ eval(f.readline()) for f in ffs ]
shuffled_files = [[fk, ff] for fk, ff in zip(first_kvs, ffs)]
shuffled_files.sort(key = lambda x:x[0][0], reverse=True)
with open(f'{self.temp_dir}/final_one.dat', mode='w') as fw:
while len(shuffled_files ) > 1:
first_keys = [x[0][0] for x in shuffled_files]
min_key = first_keys[-1]
min_idx_bound = first_keys.index(min_key)
sffs = shuffled_files[min_idx_bound:]
for sff in sffs:
fw.write(str(sff[0]) + '\n')
n = sff[1].readline()
if n :
sff[0] = eval(n)
else:
sff[0] = ''
sff[1].close()
shuffled_files = sorted(filter(lambda x:x[0], shuffled_files), key = lambda x:x[0][0], reverse=True)
if shuffled_files:
fw.write(str(shuffled_files[0][0]) + '\n') #already exist
for line in shuffled_files[0][1]:
fw.write(line)
shuffled_files[0][1].close()
def __run_reduce(self, reduce_f, outfile):
def read_mapper_output(file):
for line in file:
yield eval(line.rstrip())
with open(f'{self.temp_dir}/final_one.dat', encoding='utf-8') as f , \
open(f'{self.temp_dir}/{outfile}', encoding='utf-8',mode='w') as fw:
stdin_generator=read_mapper_output(f)
for key, kv_pairs in itertools.groupby(stdin_generator, itemgetter(0) ):
for key,result in reduce_f(key, kv_pairs):
fw.write(f"{key}\t{result}\n")
def run_mrjob(self, map_f, reduce_f, infile, outfile):
#====map
self.__run_map(map_f, infile)
#====shuffle===
self.__merge_sort_from_files()
#====reduce
self.__run_reduce(reduce_f, outfile)
def map_func(x):
for t in x.strip().split():
yield (t, 1)
def reduce_func(key , kv_pairs):
s = 0
for k,v in kv_pairs :
s += v
yield (key, s)
if __name__ == "__main__":
mr = MapReduce(30, "./")
mr.run_mrjob(map_func, reduce_func, "words.txt", "result.txt")代码明显复杂了很多，其中reduce读取有序文件还借鉴了Hadoop streaming的处理方式。
4. MapReduce思想的总结
这个系列到这算完结了，为了讲解MapReduce我都是先讲解高阶函数map和reduce，深入思考以后发现这些个东西确实还有一些值得补充的地方。但应该不止于此，对于没有完全学过函数式编程的我来说就纯个人观点一下。
写map有什么用？在你写了很多独立地循环处理逻辑以后， 你发现你可以把循环与处理分离了。反过来更多使用map会习惯这种分离的思维，甚至提升这种分离。 这种分离意味着一种共性的抽象。
reduce被设计出来的意义何在？ 不使用全局变量，而是部分变量传递，以完成全局功能的设计。全局功能线性拆分成N个独立的有部分依赖的功能总和。其本质也是一种分离或拆分思想。反过来使用reduce实现功能，习惯这种等价的整体拆分成部分的思维，强化这种拆分，意味着进行带依赖的共性的提炼。
最后概括一下MapReduce的学习意义：证明了对任务的工序拆分能有效实现并行加速。即对任务抽象拆分出同质（独立或依赖）的步骤，这些同质工作能并行/自动化。而工序拆分，靠的是前面高阶函数训练出来的思维。
