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