Python 实践之 400 行 Python 写一个类 Hadoop 的 MapReduce 框架

Google 的老三篇已经快问世十年了，但这10年里我只是肤浅的了解了一下Mapreduce的原理，并无深入的实践过程真是惭愧。最近2天，抱恙在家，开始了学习Hadoop 的工作原理。说起来Hadoop 极其复杂，各种派生类盘根错杂。但是其实MapReduce的本身工作原理却并不太复杂。可以说很好懂，简单的来说就是将数据分块，并行的进行MAP 操作，再将并行进行Reduce操作。利用并行计算的优势，减少业务的处理时间。

为此我花了2天时间，用python 实现了一套类Hadoop的Mapreduce框架。更确切的说这只是一个简单的模型， 用户只要写一个python文件，将业务的 InputSplit , Map, Reduce, OutputFormat 四个接口简单的实现。框架会根据传入的参数，依次调用这几次函数，完成一个最简单Mapreduce的程序了。

例子一共用了400 行，主要实现以下几个功能，

一个sort 的客户程序， 实现了 InputSplit , Map, Reduce, OutputFormat 这四个接口，完成具体业务相关的。 它完成的业务 就是sort –k 1 –n xxx.txt | uniq –c | awk’{print $2” “$1}’

一个server程序，它主要起到了一个消息队列的作用，主要缓存每次业务做完中间结果。
一个task_master 主程序，他是业务的灵魂，它负责启动各次业务的启动，启动多个子进程模拟多节点情况，并行地运行MAP 和Reduce操作。同时并关注每次业务完成情况，负责整体业务流程。
其他一些工具函数。比如序列化对象保存到本地文件。


从宏观的角度看MapReduce的过程 （图摘自 Hadoop技术内幕）


从图中我们可以知道split ， map （group by sort），reduce task 这几个步骤是靠客户程序来完成。
而系统则是完成整体业务，并将这几个步骤串起来 就完成了Hadoop框架本身的功能。



这个框架本身极其简陋的，不支持分布式文件系统，只能用本地文件模拟。不支持多机处理只能用本地进程模拟worker。没有单点故障监控，更不支持业务调度这样的复杂功能。
单纯但从一个业余python开发者的角度看，我认为用400来行代码就实现了一个Hadoop 的Mapreduce的原型说明python 的表现能力和效果都是非常惊人。 是实现项目原型很好的一种选择。

技术实现：
1） 动态调用python 模块
框架实现以后，就应该不会再修改了。它可以一直运行，客户只需要编写符合要求的python 模块，这个框架就能自动去调用了。
那么如何让框架去调用一个它甚至不知道的名字的模块？
比如 我要做一个wordcount 的操作， 完成代码编写以后， 我只需要将字符串 “wordcount” 传入框架，框架就会自动去执行 wordcount 类里定义的 InputSplit, Map, Reduce, OutputFormat 这样几个函数。 下一次 我又写了一个 “TOP10” 的操作，我只需要将字符串 “TOP10” 传入框架，框架就会自动去执行 TOP10类里定义的 InputSplit , Map, Reduce, OutputFormat 这样几个函数。

JAVA里我们有反射能做到这点，Python 里也有类似的方法 （GO是静态语言，我不知道如何实现，有高人指点请指点）代码如下
def run_task(modename, functionname,arg):
obj = __import__(modename) # import module
c = getattr(obj,modename)
obj = c() # new class
fun = getattr(obj,functionname)
fun(arg) # call def

自此通过参数将 类名的字符串，函数名，参数都以字符串的形式传进去，一切就都好了。

2)序列化
每个业务的中间结果都是一些MAP，但是如何传给下一个工序呢？ 我采用的方法是 使用python 序列化将中间结果对象，序列化成一个文件，将文件名保存在消息队列中，下一道工序通过消息队列获取相应的文件名，反序列化文件。获得上一次步骤的中间结果，继续操作。

3）消息队列
就是一个全局的MAP，以 任务ID 作为主键的一个分层MAP。

剩下的事情，就是耐心了。我相信一个普通的程序员几个钟头都是能完成的了。

代码在：

https://github.com/xiaojiaqi/py_hadoop