Hadoop MapReduce 实现

虽然Hadoop现在有一点过时, 但是一般的金融公司还是会用它出隔日的报表(离线计算), 本文主要关注其Map Reduce的实现.

函数编程

Map/Reduce的思想借鉴于函数式编程. Map是进行过滤和排序, 比如把一组学生按名字排序到队列, 一个名字一个队列, 然后Reduce方法进行总结, 比如对队列里的名字做统计, 得出名字出现频率. 这种思想是split-apply-combine的一种特例(见pandas groupby ).

我们知道多线程编程的局限在于访问共享资源的竞争问题, 一般需要锁, 信号量(semaphore)等技术去协调, 不然死锁等问题将会出现.

但是我们可以完全换个思路, 比如消除需要访问共享资源的限制, 这样我们就不需要锁之类的技术了.这也是函数计算的一个基本概念. 数据通过函数的参数传递, 同一时间只有一个激活的函数运行,这样就避免了冲突.

可以把函数连接作有向无环图Direct Acylic Graph, 由于函数没有隐藏的依赖, 这样多个DAG就可以并行运行.

Map/Reduce函数

Map/Reduce是一种特殊(简单)的DAG. 图如下所示: 每个map函数把一组数据按key分为key/value对, 然后不同key的元素跑到不同的计算节点, 在那里进行reduce合并.

map(input_records) {
emit(k1, v1)
...
emit(k2, v2)
...

reduce(key, values) {
aggregate = initialize()
while (values.has_next){
    aggregate = merge(values.next)
}
collect(key, aggregate)
}

可以有多个`map/reduce`组合替代一个并行的算法: ![https://s3.ap-southeast-1.amazonaws.com/kopei-public/%E5%B1%8F%E5%B9%95%E5%BF%AB%E7%85%A7%202018-10-14%20%E4%B8%8B%E5%8D%884.25.43.png](https://s3.ap-southeast-1.amazonaws.com/kopei-public/%E5%B1%8F%E5%B9%95%E5%BF%AB%E7%85%A7%202018-10-14%20%E4%B8%8B%E5%8D%884.25.43.png)

分布式文件系统(HDFS)

Hadoop需要分布式文件系统, 用于处理大文件的顺序读写.每一个大文件会被分割成块, 存储在不同数据节点.

主节点NameNode会记录所有文件的目录结构和各个块所在的位置. 主节点作为中心控制点一般会有hot standby的复制.

想要读取文件, 客户端会计算所需块在文件的偏移位置, 得出块的索引, 然后对NameNode做出请求, 然后NameNode会返回哪个DataNode有数据, 客户端就会直接和DataNode联系.

想要写入一个文件, 客户端会先和NameNode通信, 作为响应, NameNode会告诉客户端现在有哪些DataNode并且谁是主节点和哪些是从复制. 然后客户端就会把文件上传到所有DataNode, 不过DataNode这时还只会存储在buffer, 等到所有节点都存完缓存, 客户端发起commit给主节点, 主节点就会提交更新, 同时通知从节点更新, 等到所有从节点都commit, 主节点就会返回客户端提交成功. (所以DFS写是强一致性) 最后客户端还需告诉NameNode所有更新信息. 包括块分布的位置和元信息都会写入NameNode的操作日志operation log. 这个日志十分重要, 可以用于灾后恢复. NameNode也会通过不间断地checkpoint维护它的持久化状态.

当NameNode挂了, 所有的写操作将会失效, 读操作可能不受影响, 只要在客户端与DataNode的句柄有效. 需要恢复NameNode, 从节点会从上一次的checkpoint状态恢复, 并做操作日志回放.

当一个DataNode挂了, NameNode会从心跳中检查到, NameNode就会把它从集群中移除, 然后把它存储的chunk在其他节点写入. 这样做才能维护hadoop所需的replication factor.

如果这个挂掉的DataNode后来恢复了, 那么将会重新加入集群, 它会给NameNode报告所有它有的块, 每一个块是有版本号的, 所以NameNode可以检查是否这个DataNode数据是否已经过时, 如果是那么这个节点将会被后续回收.