郑州高端网站建设公司,营销型网站建设企业,株洲房地产信息网,学校网站asp源码目录
1. MapReduce概述
2. 极简MapReduce内存版
3. 复杂MapReduce磁盘版
4. MapReduce思想的总结 1. MapReduce概述 以前写过一篇 MapReduce思想 #xff0c;这次再深入一点#xff0c;简单实现一把单机内存的。MapReduce就是把它理解成高阶函数#xff0c;需要传入map和…目录
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和valuereduce时候只对一组的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 yx
剩下要做的事就是把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, keylambda 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因此设计了一个类过程就不解释了
from operator import itemgetter
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, modew) 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, moder, encodingutf-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], reverseTrue) with open(f{self.temp_dir}/final_one.dat, modew) 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], reverseTrue)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, encodingutf-8) as f , \open(f{self.temp_dir}/{outfile}, encodingutf-8,modew) as fw:stdin_generatorread_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):#mapself.__run_map(map_f, infile)#shuffleself.__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读取有序文件还借鉴了Hadoop streaming的处理方式。
4. MapReduce思想的总结 这个系列到这算完结了为了讲解MapReduce我都是先讲解高阶函数map和reduce深入思考以后发现这些个东西确实还有一些值得补充的地方。但应该不止于此对于没有完全学过函数式编程的我来说就纯个人观点一下。
写map有什么用在你写了很多独立地循环处理逻辑以后 你发现你可以把循环与处理分离了。反过来更多使用map会习惯这种分离的思维甚至提升这种分离。 这种分离意味着一种共性的抽象。
reduce被设计出来的意义何在 不使用全局变量而是部分变量传递以完成全局功能的设计。全局功能线性拆分成N个独立的有部分依赖的功能总和。其本质也是一种分离或拆分思想。反过来使用reduce实现功能习惯这种等价的整体拆分成部分的思维强化这种拆分意味着进行带依赖的共性的提炼。
最后概括一下MapReduce的学习意义证明了对任务的工序拆分能有效实现并行加速。即对任务抽象拆分出同质独立或依赖的步骤这些同质工作能并行/自动化。而工序拆分靠的是前面高阶函数训练出来的思维。
