首页 > 代码库 > 二、基于hadoop的nginx访问日志分析---计算日pv
二、基于hadoop的nginx访问日志分析---计算日pv
代码:
# pv_day.py
#!/usr/bin/env python # coding=utf-8 from mrjob.job import MRJob from nginx_accesslog_parser import NginxLineParser class PvDay(MRJob): nginx_line_parser = NginxLineParser() def mapper(self, _, line): self.nginx_line_parser.parse(line) day, _ = str(self.nginx_line_parser.time_local).split() yield day, 1 # 每一天的 def reducer(self, key, values): yield key, sum(values) def main(): PvDay.run() if __name__ == ‘__main__‘: main()
代码解释:
定义了一个集成MRJob类的job类,这个类包含定义好的steps。
一个‘step’包含一个mapper,combiner和一个reducer,这些是可选的,但是必须使用至少一个。
mapper()方法有两个参数key,value(这个例子中,key被忽略,每行日志作为一个value),并生成key-value对。
reduce()方法接受一个key和一个可迭代的values,并生成许多key-value对(这个例子中,计算每个key对应values值得和,也就是每天对应的PV)。
以不同方式执行job:
基本方式:
# python3 pv_day.py access_all.log-20161227
No configs found; falling back on auto-configuration
Creating temp directory /tmp/pv_day.root.20161228.022837.113256
Running step 1 of 1...
Streaming final output from /tmp/pv_day.root.20161228.022837.113256/output...
"2016-12-27" 47783
"2016-12-26" 299427
Removing temp directory /tmp/pv_day.root.20161228.022837.113256...
标准输入stdin方式,这种方式只接受第一个文件
# python3 pv_day.py < access_all.log-20161227 No configs found; falling back on auto-configuration Creating temp directory /tmp/pv_day.root.20161228.024431.884434 Running step 1 of 1... reading from STDIN Streaming final output from /tmp/pv_day.root.20161228.024431.884434/output... "2016-12-27" 47783 "2016-12-26" 299427 Removing temp directory /tmp/pv_day.root.20161228.024431.884434...
混合方式:
python3 pv_day.py input1.txt input2.txt - < input3.txt
分布式:
默认情况下,mrjob执行job使用单个Python进程,这里只是调试,并不是精确的分布式计算!
如果使用分布式计算,可以使用 -r/--runner选项。 使用 -r inline(默认), -r local, -r hadoop, -r emr
# python pv_day.py -r hadoop hdfs://my_home/input.txt
二、基于hadoop的nginx访问日志分析---计算日pv
声明:以上内容来自用户投稿及互联网公开渠道收集整理发布,本网站不拥有所有权,未作人工编辑处理,也不承担相关法律责任,若内容有误或涉及侵权可进行投诉: 投诉/举报 工作人员会在5个工作日内联系你,一经查实,本站将立刻删除涉嫌侵权内容。