Hadoop Mapreduce Kpi 用Hadoop提取KPI统计指标

用Hadoop提取KPI统计指标

前言

Web日志包含着网站最重要的信息，通过日志分析，我们可以知道网站的访问量，哪个网页访问人数最多，哪个网页最有价值等。一般中型的网站(10W的PV以上)，每天会产生1G以上Web日志文件。大型或超大型的网站，可能每小时就会产生10G的数据量。对于日志的这种规模的数据，用Hadoop进行日志分析，是最适合不过的了。

WEB日志概述

Web日志由Web服务器产生，可能是Nginx, Apache, Tomcat等。从Web日志中，我们可以获取网站每类页面的PV值（PageView，页面访问量）、独立IP数；稍微复杂一些的，可以计算得出用户所检索的关键词排行榜、用户停留时间最高的页面等；更复杂的，构建广告点击模型、分析用户行为特征等等。在Web日志中，每条日志通常代表着用户的一次访问行为，例如下面就是一条nginx日志：

222.68.172.190 - - [18/Sep/2013:06:49:57 +0000] "GET /images/my.jpg HTTP/1.1" 200 19939
"http://www.angularjs.cn/A00n" "Mozilla/5.0 (Windows NT 6.1)
AppleWebKit/537.36 (KHTML, like Gecko) Chrome/29.0.1547.66 Safari/537.36"

拆解为以下8个变量
remote_addr: 记录客户端的ip地址, 222.68.172.190
remote_user: 记录客户端用户名称, –
time_local: 记录访问时间与时区, [18/Sep/2013:06:49:57 +0000]
request: 记录请求的url与http协议, “GET /images/my.jpg HTTP/1.1″
status: 记录请求状态,成功是200, 200
body_bytes_sent: 记录发送给客户端文件主体内容大小, 19939
http_referer: 用来记录从那个页面链接访问过来的, “http://www.angularjs.cn/A00n”
http_user_agent: 记录客户浏览器的相关信息, “Mozilla/5.0 (Windows NT 6.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/29.0.1547.66 Safari/537.36″

KPI指标设计

PV(PageView): 页面访问量统计

Map过程{key:$request,value:1}
Reduce过程{key:$request,value:求和(sum)}

IP: 页面独立IP的访问量统计

Map: {key:$request,value:$remote_addr}
Reduce: {key:$request,value:去重再求和(sum(unique))}

Time: 用户每小时PV的统计

Map: {key:$time_local,value:1}
Reduce: {key:$time_local,value:求和(sum)}

Browser: 用户的访问设备统计

Map: {key:$http_user_agent,value:1}
Reduce: {key:$http_user_agent,value:求和(sum)}

Source: 用户来源域名的统计

Map: {key:$http_referer,value:1}
Reduce: {key:$http_referer,value:求和(sum)}

架构

日志是由业务系统产生的，我们可以设置web服务器每天产生一个新的目录，目录下面会产生多个日志文件，每个日志文件64M。 设置系统定时器CRON，夜间在0点后，向HDFS导入昨天的日志文件。 完成导入后，设置系统定时器，启动MapReduce程序，提取并计算统计指标。 完成计算后，设置系统定时器，从HDFS导出统计指标数据到数据库，方便以后的即使查询。

程序开发

public class KPIIP {

public static class KPIIPMapper extends MapReduceBase implements Mapper<Object, Text, Text, Text> {
private Text word = new Text();
private Text ips = new Text();

@Override
public void map(Object key, Text value, OutputCollector<Text, Text> output, Reporter reporter) throws IOException {
KPI kpi = KPI.filterIPs(value.toString());
if (kpi.isValid()) {
word.set(kpi.getRequest());
ips.set(kpi.getRemote_addr());
output.collect(word, ips);
}
}
}

public static class KPIIPReducer extends MapReduceBase implements Reducer<Text, Text, Text, Text> {
private Text result = new Text();
private Set<String> count = new HashSet<String>();

@Override
public void reduce(Text key, Iterator<Text> values, OutputCollector<Text, Text> output, Reporter reporter) throws IOException {
while (values.hasNext()) {
count.add(values.next().toString());
}
result.set(String.valueOf(count.size()));
output.collect(key, result);
}
}

public static void main(String[] args) throws Exception {
String input = "hdfs://192.168.1.210:9000/user/hdfs/log_kpi/";
String output = "hdfs://192.168.1.210:9000/user/hdfs/log_kpi/ip";

JobConf conf = new JobConf(KPIIP.class);
conf.setJobName("KPIIP");
conf.addResource("classpath:/hadoop/core-site.xml");
conf.addResource("classpath:/hadoop/hdfs-site.xml");
conf.addResource("classpath:/hadoop/mapred-site.xml");

conf.setMapOutputKeyClass(Text.class);
conf.setMapOutputValueClass(Text.class);

conf.setOutputKeyClass(Text.class);
conf.setOutputValueClass(Text.class);

conf.setMapperClass(KPIIPMapper.class);
conf.setCombinerClass(KPIIPReducer.class);
conf.setReducerClass(KPIIPReducer.class);

conf.setInputFormat(TextInputFormat.class);
conf.setOutputFormat(TextOutputFormat.class);

FileInputFormat.setInputPaths(conf, new Path(input));
FileOutputFormat.setOutputPath(conf, new Path(output));

JobClient.runJob(conf);
System.exit(0);
}

}

文章来源:http://pigpdong.github.io/hadoop/2014/05/10/Hadoop-Mapreduce-KPI/