极简的MapReduce实现

目录

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的内置函数，中间分组也容易用sort+groupby实现。因此首先约定map和reduce的自定义函数写法就是直接按内置函数要求来写，只不过map输出是键值对，要求用长度是2的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_result

3. 复杂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 = 0for k,v in kv_pairs :   s += vyield (key, s)

很显然我们要补充很多东西，要读取文件、要设计缓存、要做merge，因此设计了一个类，过程就不解释了

import os
import itertools
import sys
class MapReduce:class Buffer:def __init__(self, buffer_size = 80000):self.buffer_size = buffer_sizeself.buffer = []self.current_size = 0def 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 = 0def __init__(self, buffer_size,temp_dir):self.buffer = self.Buffer(buffer_size)self.temp_dir = temp_dirself.block_id = 0def __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 += 1self.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 existfor 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, lambda x:x[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):#====mapself.__run_map(map_f, infile)#====shuffle===self.__merge_sort_from_files()#====reduceself.__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 = 0for k,v in kv_pairs :   s += vyield (key, s)if __name__ == "__main__":mr = MapReduce(30, "./")mr.run_mrjob(map_func, reduce_func, "words.txt", "result.txt")

代码明显复杂了很多，其中reduce读取有序文件依次分key的组，还借鉴了Hadoop streaming 中Python代码的处理方式。

4. MapReduce思想的总结

       这个系列到这算完结了，为了讲解MapReduce我都是先讲解高阶函数map和reduce，深入思考以后发现这些个东西确实还有一些值得补充的地方。但应该不止于此，对于没有完全学过函数式编程的我来说就纯个人观点一下：

写map有什么用？在你写了很多独立地循环处理逻辑以后， 你发现你可以把循环与处理分离了。反过来更多使用map会习惯这种分离的思维，甚至提升这种分离。 这种分离意味着一种共性的抽象。

reduce被设计出来的意义何在？ 不使用全局变量，而是部分变量传递，以完成全局功能的设计。全局功能线性拆分成N个独立的有部分依赖的功能总和。其本质也是一种分离或拆分思想。反过来使用reduce实现功能，习惯这种等价的整体拆分成部分的思维，强化这种拆分，意味着进行带依赖的共性的提炼。

最后概括一下MapReduce的学习意义：证明了对任务的工序拆分能有效实现并行加速。即对任务抽象拆分出同质（独立或依赖）的步骤，这些同质工作能并行/自动化。而工序拆分，靠的是前面高阶函数训练出来的思维。