[转]Hadoop之MapReduce快速入门

最近有需要分析某个省份的的dns日志,从中解析出一段时间top100的域名排行,日志量一天大概有3、4G的样子,数据量还是很客观。好消息是公司已经有了完整的hadoop集群,借助hadoop分析还是很方便的,坏处是我一点都不懂hadoop,囧个。不过之前写过类似的分析dns日志脚步算是有点经验了,总之硬着头皮上吧,开搞。

1.hadoop基本操作

磨刀不误砍柴工,纵身登上hadoop,快速的敲了个history偷看了下同事的操作记录,看到了最常用的几条指令,心里有底了。区区3、4条命令,小样,我真不信我搞不了你。

hadoop fs -ls /user/hadoop/

hadoop fs -rmr /user/hadoop/sicong

hadoop fs -get /user/hadoop/sicong/part-00000 ./result/

大致有如上几条,稍微解释下:

a)以上所有指令都是使用hadoop用户执行的,so首先需要切换到hadoop用户 su – hadoop,我第一次就不知道土鳖了一把

b)hadoop 所有的文件系统操作都是以hadoop fs 开头的,hadoop fs -ls 即对执行目录执行ls操作,作用等同Linux下的ls,hadoop fs -rmr 删除执行的目录,很危险,千万慎重

c)hadoop采用的是分布式的HDFS,彼此节点可以相互通信,因为登录上hadoop集群任一节点执行的操作都会在全集群执行的,so 你懂的,小心~

2.MapReduce简介:

前面稍微介绍了基本的hadoop操作,下面要说的就是大名鼎鼎的MapReduce.

1.什么是mapreduce:

mapreduce是hadoop最小的工作单元,其实分为两个处理阶段,map阶段、reduce阶段,当然中间可能还有Shuffle阶段。每个阶段都是以键/值的形式作为输入,输出,同时hadoop的streaming采用了系统标准的输入、输出,更方便我们的使用,可以使用任何语言编写map脚本、reduce脚本。

2.为什么要使用mapreduce:

小规模的数据处理我们当然可以使用系统的awk、python脚本等进行处理的,首先我们就用python脚本处理。具体的Python脚本以及脚本思路参加前面文章:python脚本分析dns日志获取域名排行。

但可以看出前篇文章的脚本程序、原始数据都是在本机的,这个脚本在处理小规模数据(百M级别)的毫无问题,但遇到大规律的数据就无能为力了,受制于单机的计算能力、内存、存储。当数据规模达到T级别的时候本地的存储就无法承受了,数据会分别在不同的机器甚至机架上,前面的脚本就只能利用本地存储、本机内存,根本没办法处理。

hadoop基于HDFS的分布式存储将数据存储到大量的hadoop节点上,处理的时候目标数据会被拆分了等长的小数据块(分片),同时在不同的节点启动任务处理相应的数据块,之后排序,将结果合并、汇总,交由reduce处理。这个过程中充分利用了集群的计算能力、内存,同时将数据拆分,由统一的存储、任务调度等,so这就是mapreduce的意义。

3.mapreduce数据流:

上图是hadoop权威指南的一个插图,较好的解释了map阶段、reduce阶段的数据流向。

3.必要的准备:

开始动手之前大致说下必要的准备,首先建立一个目录,在该目录下存放你相应的目标样本文件,以及相应的mapper.py 、reducer.py等。

[[email protected] sicong]$ ls
domain.py  log  mapper.py  queries.log.CMN-CQ.20130830031330.gz  reducer.py  reducer.py.bak  result  startup.sh

我的目录结构如上,queries.log.CMN-CQ.20130830031330.gz为目标文件的一个样本,用来测试的,result用以存储最后的输出结果。

startup.sh是抄了同事(xdays,大神一枚)的用以简化hadoop的操作:

[[email protected] sicong]$ cat startup.sh

#!/bin/sh

wd=/home/hadoop/sicong/

# 1. get raw result sorted by domain and count
hadoop jar \
$HADOOP_HOME/contrib/streaming/hadoop-streaming-1.0.3.jar \
-file ${wd}mapper.py -mapper ${wd}mapper.py \
-file ${wd}reducer.py -reducer ${wd}reducer.py \
-input /user/hadoop/fdns_logs/08/30/CMN-CQ-2-*  \
-output /user/hadoop/sicong/  \
-jobconf stream.non.zero.exit.is.failure=false \
-jobconf stream.recordreader.compression=gzip \
-jobconf mapred.map.max.attempts="3″ \
-jobconf mapred.reduce.max.attempts="3″ \
-jobconf mapred.max.map.failures.percent="1″

wd参数为你建立的目录,-input 执行的你目标数据文件,-output 指定数据结果存储路径,咳咳其实这段解释也是抄同事的。但需要注意的是这里的input、output都是hadoop的目录。

4.编写mapreduce

[[email protected] sicong]$ cat mapper.py
[sourcecode language="python"]
#!/usr/bin/env python
import sys
for line in sys.stdin:
line = line.strip()
segs = line.split('|')
print '%s' % segs[2]

[[email protected] sicong]$ cat reducer.py
#!/usr/bin/env python

from operator import itemgetter
import sys

domain_list = {}

for key in sys.stdin:
key = key.strip()

if key in domain_list:
domain_list[key] += 1
else:
domain_list[key] = 1

count = 0
for v in sorted(domain_list.iteritems(),key =lambda x:x[1],reverse=True):
print v[1],v[0]
#display the top 100
if count > 100:
break
count += 1

可以看出没什么特别的,只是将前面的domain.py文件拆分了,将数据提前的部分拆分到map.py中,将数据排序、计算的部分拆分到reduce.py,当然其实也可以不拆分,在一个脚本里面写了,只是基于性能的问题拆分了。

5.测试mapreduce:

写好了别着急跑脚本,最好先测试下,明白了hadoop的mapreduce过程就可以用如下命令去模仿mapreduce的流程去测试下:

如果执行成功了,则表示mapreduce脚本无大的问题,如果失败,那。。你就慢慢该脚本吧。

5.执行

真到可以其实就很简单了,直接拍下脚本就好了,sh startup.sh 就好了。

最后别忘了把输出结果(默认为part-00000)取回来,因为output指定的目录为hadoop的目录,因此需要将结果从hadoop中拷贝到系统目录下(我保存在3中目录下的result下):

hadoop dfs -get /user/hadoop/sicong/part-00000 ./result/

6.总结: mapreduce不难写,难在写好一个mapreduce,如果中间采用了正则匹配等很耗时间的操作,那很有可能mapreduce会跑失败。同样这个脚步也有很多可以优化的地方,例如最后的reduce阶段可以不排序直接输出,然后进入下一个mapreduce使用hadoop进行排序;或者每次map阶段之后就可以进行计数、排序之后再进入reduce阶段。总之尽可能的拆分任务,加快任务的处理。