极简的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 = 0
    for k,v in kv_pairs :   
        s += v
    yield (key, s)

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

import os
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, 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):
        #====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读取有序文件依次分key的组，还借鉴了Hadoop streaming 中Python代码的处理方式。

4. MapReduce思想的总结

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

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

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

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