你好,游客 登录
背景:
阅读新闻

深入浅出Hadoop之mapreduce

[日期:2018-02-23] 来源:  作者:卿哥聊技术 [字体: ]

之前已经作出预告,那么今天就聊聊mapreduce,起源于Google的map reduce paper, 而后经历了mapreduce 1,和构建于yarn上的mapreduce 2,mapreduce1 除了提供一定的历史演变价值和了解一下mapreduce最初的设计之外就没有必要学了哈,毕竟现在意义上的mapreduce2,spark都是在yarn上。当然mapreduce这项技术本身可能现在也是逐年衰落,不是我说的,是michael stonebraker3年前就说了(stonebraker认为程序员只需要了解sql就行了,nosql啥的都应该直接或间接支持用sql来查询交互),而且google自己也早就不用了,不过mapreduce还是有自身一定的学习价值,比如map, combiner, shuffle/sort, practiioner, reducer,消息传递, data locality(即把运算移动到数据旁,而不是传输数据来节省网络带宽提高运算效率)都是分布式系统运算框架的一个里程碑。很多分布式系统设计课程比如MIT研究生著名的分布式系统设计的前几章必然要讲mapreduce。

分布式系统前言

分布式系统由于包括很多node,所以它的根基是unreliable component包括node,network和clock,上层的设计必然需要考虑到这一点。Unreliable node有如下三种表现形式:

  • fail-stop, 比如电力供应中断了(比如data center停电了,或者地震了发水灾了啥的),fail了就完了,无法恢复。
  • fail-recovery,比如node升级kernel,os,software,需要重启
  • 拜占庭 failure,这个比较狠,就是这个node看似正常,但是你说城门楼子它说胯骨轴子,你说往东,它往西走。相当于一个神经错乱的node或者说是被入侵了的node。

同时分布式系统涉及大量的网络传输,不管是RPC还是RESTFUL都是走网络,网络也是unreliable的,有如下三种表现形式:

  • 完美传输,即%0 loss, 100% in order,俗称happy home,这个资源成本比较高,适用于特别重要的服务
  • fair-loss, 这个最为常见,就是正常丢包,不已传输内容为转移。TCP/IP就是为了解决这个事儿而设计了3 way handshake,retry,sliding window,congestion control啥的
  • 拜占庭 failure,这个就基本相当于被man in the middle 了,anything is possible,good luck :)思考题:SSL/TLS 能有效解决man in the middle吗?

还没完,分布式系统还有一个问题就是clock,一方面每台机器的时间都可能不一样,俗称clock skew,另一方面,每台机器对每一秒的感知也不一样,俗称clock drift。所以伟大的lamport(2013图灵奖得主,分布式大神,latex,vector clock,paxos,etc)发明了logical clock,其中最著名的是vector clock。相当于让每个event有了自己的先后顺序。这个可以单独聊一次它的具体原理。插一句嘴,我觉得分布式系统这么多年敢称大神的只有两个,理论大师lamport和实战天王Jeff Dean(Jeff Dean当年在MIT淡定的给我们介绍谷歌分布式系统设计经验,然后说自己一个周末现学maching learning,搞出了个猫图片识别,当时我就在想他学machine learning干啥,然后tensor flow这个项目就横空出世了。。。orz)

还有就是分布式系统实现分为synchronous和asynchronous两种model,synchronous就是blocking callback with optional timeout,asynchronous就是event call back with optional timeout。

下面我们把以上几种情形组合一下:

  • fail-stop + 完美传输 + synchronous,比如超级计算机每个processor由local high speed bus相连,user case是OpenMP 和 MPI
  • fail-recovery + fair-loss + asynchronous,就是我们最近一直聊的hadoop eco system了
  • 拜占庭 node + 拜占庭网络+asynchronous,这就是分布于untrusted computer和untrusted network之中的grid computing了

map

先看一下python的map function,接下来会聊hadoop streaming

>>> map(lambda x: x*x, [1,2,3,4,5])
[1, 4, 9, 16, 25]

or>>> items = (1,2,3,4,5)

>>> def sqr(x): return x**2>>> map(sqr,items)

[1, 4, 9, 16, 25]

reduce

先看一下python的reduce function

>>> reduce(operator.iadd, [1,4,9,16,25])
55

or

>>> reduce(lambda x,y: x+y, [1,4,9,16,25])
55

data flow

map 和 reduce的input/output都是key/value pair。注意map或者reduce不一定都需要,比如grep,map=grep,reduce=None

在sort和shuffling阶段,sorting用的是external sorting,所以不用担心内存爆了。

实现

上图中,intermidiate result是存放在local disk中而不是HDFS,因为就算丢了,也可以通过map重新得到,所以不用使用HDFS做multiple copy。

上图可以看出yarn分为resource manager和node manager,resource manager会launch application master,application master会请求resource根据resource富余程度launch application process。

下面这张图也是这个意思:

具体来说,input data会被Hadoop切割为固定大小的input splits,Hadoop 会为每个split creates一个map task,map task会对split中的每一个record运行user-defined map function. 对于大部分job来说,a good split size是一个HDFS block,128MB。Hadoop尽量通过data locality optimization来让map task运行在存有input data的节点上。如果不行,就选择同一个rack上的其它node,如果还不行,就选择旁边rack上的node。

reduce就没有data locality一说了,input都是通过网络传过去的。这中间还会有一个shuffle&&sort的过程,通常就是通过一个hash function来把相同的key放在一起,保证对于每一个given key来说,所有的records都在一个partition里。所以当reducer处理的时候可以保证input data都是按照key sorting好的了。

Hadoop Streaming

首先,说说distributed cache, 比如你有个文件需要在运算的时候用到如何传给map/reduce task呢?答,是通过-files/-archives/-libjars 传过去的,比如-files mapper.py, reducer.py, some_file.txt, 此时some_file.txt会被传到需要的node上,每个node只需要一个copy,注意some_files.txt是read only的,所以可以被tasks共享。 archives是打包传输,libjars是传jar格式。

following method 可以创建archive文件:

tar -cf pack.tar a.txt b.txt c.txt

读的时候就用如下path:

pack.tar/a.txt,pack.tar/b.txt, pack.tar/c.txt

其次,说说environment variable, hadoop streaming可以通过-D some_var="some value"的方式把变量传给nodes。

第三,从task的角度可以通过reporter:status: 的方式把report传回去。

下面举个word count的实际例子,比如统计wikipedia的word count:

mapper.py

#!/usr/bin/python

import sys
import re

reload(sys)
sys.setdefaultencoding('utf-8') # required to convert to unicode

for line in sys.stdin:
    try:
        article_id, text = unicode(line.strip()).split('\t', 1)
    except ValueError as e:
        continue
    words = re.split("\W*\s+\W*", text, flags=re.UNICODE)
    for word in words:
        print "%s\t%d" % (word.lower(), 1)

reducer.py

#!/usr/bin/python

import sys

current_key = None
word_sum = 0

for line in sys.stdin:
    try:
        key, count = line.strip().split('\t', 1)
        count = int(count)
    except ValueError as e:
        continue
    if current_key != key:
        if current_key:
            print "%s\t%d" % (current_key, word_sum)
        word_sum = 0
        current_key = key
    word_sum += count

if current_key:
    print "%s\t%d" % (current_key, word_sum)

本机运行:

cat wiki.txt | ./mapper.py | sort | ./reducer.py

Hadoop streaming:

OUT_DIR="wiki_wordcount_result_"$(date +"%s%6N")
NUM_REDUCERS=8

hdfs dfs -rm -r -skipTrash ${OUT_DIR} > /dev/null

yarn jar /opt/cloudera/parcels/CDH/lib/hadoop-mapreduce/hadoop-streaming.jar \
    -D mapred.jab.name="Streaming wordCount" \
    -D mapreduce.job.reduces=${NUM_REDUCERS} \
    -files mapper.py,reducer.py \
    -mapper "python mapper.py" \
    -combiner "python reducer.py" \
    -reducer "python reducer.py" \
    -input /wiki/en_articles_part \
    -output ${OUT_DIR} > /dev/null

好了,今天就写到这,happy mapreduce!





收藏 推荐 打印 | 录入:Cstor | 阅读:
本文评论   查看全部评论 (0)
表情: 表情 姓名: 字数
点评:
       
评论声明
  • 尊重网上道德,遵守中华人民共和国的各项有关法律法规
  • 承担一切因您的行为而直接或间接导致的民事或刑事法律责任
  • 本站管理人员有权保留或删除其管辖留言中的任意内容
  • 本站有权在网站内转载或引用您的评论
  • 参与本评论即表明您已经阅读并接受上述条款