Python实现用Hadoop的map/reduce对web日志进行统计
日志格式
61.160.241.107 - - [23/Aug/2011:22:00:00 +0800] "GET /map.php?gid=38&sid=75&user=14717213&roleid=490711&time=1314108000&user_yx=736959&levafee11f0d1bacbfecbb631192 HTTP/1.1" 200 5 "-" "Java/1.6.0_23"
以对IP 的访问量进行统计为例
map脚本map.py 清洗日志数据
#!/usr/bin/python
import sys
import re
debug = False#设置lzo文件偏移位
if debug:
lzo = 0
else:
lzo = 1
for line in sys.stdin:
ipaddress=re.compile(r'([\d.]*) (- - \[[^[\]]*\] "[^ ]* /)([^ ]*)([^ ]*\.php\?)([^ ]*)')
match=ipaddress.match(line.split('\t',1)[lzo])
if match:
ip=match.group(1)
#tb=match.group(2)
#url=match.group(4)
print ip
reduce脚本red.py 对ip数进行统计
#!/usr/bin/python
#-*-coding:UTF-8 -*-
import sys
import os
import string
res = {}
for line in sys.stdin:
skey=line[0:-1]
if(res.has_key(skey)==False):
res[skey]=0
res[skey]=res[skey]+1
for key in res.keys():
print key+"\t"+str(res[key])
将map.py red.py脚本修改为可执行权限
# chmod +x *.py
在shell中进行调试
# cat 1.log|/home/map.py|/home/red.py
在hadoop streaming中执行mapreduce作业
# hadoop jar /opt/mapr/hadoop/hadoop-0.20.2/contrib/streaming/hadoop-0.20.2-dev-streaming.jar -file /home/map.py -file /home/red.py -mapper /home/map.py -reducer /homej/red.py -input /test/a.log.-output /test/test2
其中input 与output 路径都为hdfs文件系统路径 如果将输出文件进行压缩的话需要加参数 -jobconf mapred.output.compress=true -jobconf mapred.output.compression.codec=com.hadoop.compression.lzo.LzopCodec
如果输入文件格式为lzo压缩格式的话 需要指定参数-inputformat com.hadoop.mapred.DeprecatedLzoTextInputFormat 注意的是在hadoop中使用lzo的话，偏移位需要加1。