利用Hadoop Mapreduce实现pv统计分析

摘 要

本文将介绍通过Hadoop Mapreduce实现离线统计网站每日pv的思路及代码。

前言

利用网站的kpi数据来分析出网站潜在的价值，那么了解网站的PV、UV、IP的状况，是一项必不可少的任务。本文将介绍通过Hadoop Mapreduce实现离线统计网站每日pv的思路及代码。

什么是PV

pv是指页面的浏览量或点击量（Page View），用户每访问一次或刷新一下即被计算一次。

需求

对网站以往的访问数据进行日pv、月PV、年PV统计。

技术选型

对于访问量大的网站来说，普通程序计算实现成本非常大。我们可以利用Hadoop来实现分布式计算，将固有的数据量分散到多台机器进行计算，无疑加快了计算速度，也降低了宕机的风险。

 

实现思路

在map阶段将数据清洗，并进行排序、分组。在reduce阶段完成统计。非常简单。

下面具体看实现代码：

KPI.java 用于封装日志信息及排序

  1. package com.itunic.mr.kpi;
  3. import java.io.DataInput;
  4. import java.io.DataOutput;
  5. import java.io.IOException;
  6. import java.text.ParseException;
  7. import java.text.SimpleDateFormat;
  8. import java.util.Date;
  9. import java.util.HashSet;
  10. import java.util.Locale;
  11. import java.util.Set;
  13. import org.apache.hadoop.io.WritableComparable;
  14. /**
  15.  * 封装一个实体类，将nginx的日志分解成有用的信息。
  16.  * @author itunic
  17.  *
  18.  */
  19. public class KPI implements WritableComparable<KPI> {
  20.     private String remote_addr;// 来访ip
  21.     private String remote_user;// 来访用户名称，忽略属性“-”
  22.     private String time_local;// 记录时间与时区
  23.     private String request;// 访问页面
  24.     private String status;// 返回状态
  25.     private String body_bytes_sent;// 返回客户端内容主体大小
  26.     private String http_referer;// 来访页面
  27.     private String http_user_agent;// 客户浏览器的相关信息
  29.     private boolean valid = true;// 检验数据是否合法
  30.     //设置需要统计的页面类型
  31.     static Set<String> pages = null;
  32.     static {
  33.         pages = new HashSet<String>();
  34.         pages.add(".php");
  35.         /*pages.add(".jsp");
  36.         pages.add(".png");*/
  37.     }
  39.     /**
  40.      * 将传过来的数据封装成bean
  41.      * 
  42.      * @param line
  43.      * @return
  44.      */
  45.     private static KPI parser(String line) {
  46.         KPI kpi = new KPI();
  47.         String[] arr = line.split(" ");
  48.         if (arr.length > 11) {
  49.             kpi.setRemote_addr(arr[0]);
  50.             kpi.setRemote_user(arr[1]);
  51.             kpi.setTime_local(arr[3].substring(1));
  52.             String url = arr[6].indexOf("?") != -1 ? arr[6].substring(0, arr[6].indexOf("?")) : arr[6];
  53.             kpi.setRequest(url);
  54.             kpi.setStatus(arr[8]);
  55.             kpi.setBody_bytes_sent(arr[9]);
  56.             kpi.setHttp_referer(arr[10]);
  58.             if (arr.length > 12) {
  59.                 kpi.setHttp_user_agent(arr[11] + " " + arr[12]);
  60.             } else {
  61.                 kpi.setHttp_user_agent(arr[11]);
  62.             }
  64.             if (Integer.parseInt(kpi.getStatus()) >= 400) {
  65.                 kpi.setValid(false);
  66.             }
  67.         } else {
  68.             kpi.setValid(false);
  69.         }
  70.         return kpi;
  71.     }
  73.     /**
  74.      * 提取需要的url
  75.      * 
  76.      * @param line
  77.      * @return
  78.      */
  79.     public static KPI filterPVs(String line) {
  80.         KPI kpi = parser(line);
  81.         kpi.setValid(false);
  82.         for (String page : pages) {
  83.             if (kpi.getRequest() != null) {
  84.                 if (kpi.getRequest().contains(page)) {
  85.                     kpi.setValid(true);
  86.                     break;
  87.                 }
  88.             }
  89.         }
  90.         return kpi;
  91.     }
  93.     public String getRemote_addr() {
  94.         return remote_addr;
  95.     }
  97.     public void setRemote_addr(String remote_addr) {
  98.         this.remote_addr = remote_addr;
  99.     }
  101.     public String getRemote_user() {
  102.         return remote_user;
  103.     }
  105.     public void setRemote_user(String remote_user) {
  106.         this.remote_user = remote_user;
  107.     }
  109.     public String getTime_local() {
  110.         return time_local;
  111.     }
  113.     public Date getTime_local_Date() {
  114.         SimpleDateFormat df = new SimpleDateFormat("dd/MMM/yyyy:HH:mm:ss", Locale.US);
  115.         try {
  116.             return df.parse(this.time_local);
  117.         } catch (ParseException e) {
  118.             return null;
  119.         }
  121.     }
  123.     public String getTime_local_day() throws ParseException {
  124.         SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd");
  125.         return df.format(this.getTime_local_Date());
  126.     }
  128.     public String getTime_local_Date_hour() throws ParseException {
  129.         SimpleDateFormat df = new SimpleDateFormat("yyyyMMddHH");
  130.         return df.format(this.getTime_local_Date());
  131.     }
  133.     public void setTime_local(String time_local) {
  134.         this.time_local = time_local;
  135.     }
  137.     public String getRequest() {
  138.         return request;
  139.     }
  141.     public void setRequest(String request) {
  142.         this.request = request;
  143.     }
  145.     public String getStatus() {
  146.         return status;
  147.     }
  149.     public void setStatus(String status) {
  150.         this.status = status;
  151.     }
  153.     public String getBody_bytes_sent() {
  154.         return body_bytes_sent;
  155.     }
  157.     public void setBody_bytes_sent(String body_bytes_sent) {
  158.         this.body_bytes_sent = body_bytes_sent;
  159.     }
  161.     public String getHttp_referer() {
  162.         return http_referer;
  163.     }
  166.     public void setHttp_referer(String http_referer) {
  167.         this.http_referer = http_referer;
  168.     }
  170.     public String getHttp_user_agent() {
  171.         return http_user_agent;
  172.     }
  174.     public void setHttp_user_agent(String http_user_agent) {
  175.         this.http_user_agent = http_user_agent;
  176.     }
  178.     public boolean isValid() {
  179.         return valid;
  180.     }
  182.     public void setValid(boolean valid) {
  183.         this.valid = valid;
  184.     }
  187.     @Override
  188.     public void write(DataOutput out) throws IOException {
  189.         out.writeUTF(this.getRemote_addr());
  190.         out.writeUTF(this.getRemote_user());
  191.         out.writeUTF(this.getTime_local());
  192.         out.writeUTF(this.getRequest());
  193.         out.writeUTF(this.getStatus());
  194.         out.writeUTF(this.getBody_bytes_sent());
  195.         out.writeUTF(this.getHttp_referer());
  197.     }
  199.     @Override
  200.     public void readFields(DataInput in) throws IOException {
  201.         this.setRemote_addr(in.readUTF());
  202.         this.setRemote_user(in.readUTF());
  203.         this.setTime_local(in.readUTF());
  204.         this.setRequest(in.readUTF());
  205.         this.setStatus(in.readUTF());
  206.         this.setBody_bytes_sent(in.readUTF());
  207.         this.setHttp_referer(in.readUTF());
  209.     }
  211.     /**
  212.      * 排序，按照日期降序、url升序处理。
  213.      */
  214.     @Override
  215.     public int compareTo(KPI o) {
  216.         int i;
  217.         try {
  218.             i = this.getTime_local_day().compareTo(o.getTime_local_day());
  220.             if (i != 0) {
  221.                 return -i;
  222.             }
  223.         } catch (ParseException e) {
  224.             e.printStackTrace();
  225.         }
  226.         return this.getRequest().compareTo(o.getRequest());
  227.     }
  228. }

PageViewGroup.java 用于将日志信息分组

  1. package com.itunic.mr.kpi;
  3. import java.text.ParseException;
  5. import org.apache.hadoop.io.WritableComparable;
  6. import org.apache.hadoop.io.WritableComparator;
  8. /**
  9.  * 分组，
  10.  * 由于需要按照每日的维度统计pv，必须将日期分组。
  11.  * @author itunic
  12.  *
  13.  */
  14. public class PageViewGroup extends WritableComparator {
  15.     public PageViewGroup() {
  16.         super(KPI.class, true);
  17.     }
  19.     @SuppressWarnings("rawtypes")
  20.     @Override
  21.     public int compare(WritableComparable a, WritableComparable b) {
  22.         KPI kpi1 = (KPI) a;
  23.         KPI kpi2 = (KPI) b;
  24.         try {
  25.             /**
  26.              * 判断当前bean与传入的bean的日期是否一致。如果一致则需要判断是否为同一个url
  27.              */
  28.             int i = kpi1.getTime_local_day().compareTo(kpi2.getTime_local_day());
  29.             if (i != 0) {
  30.                 return -i;
  31.             }
  32.             return kpi1.getRequest().compareTo(kpi2.getRequest());
  33.         } catch (ParseException e) {
  34.             // TODO Auto-generated catch block
  35.             e.printStackTrace();
  36.         }
  37.         return 0;
  38.     }
  39. }

PageView.java Mapreduce计算类

  1. package com.itunic.mr.kpi;
  3. import java.io.IOException;
  4. import java.text.ParseException;
  6. import org.apache.hadoop.conf.Configuration;
  7. import org.apache.hadoop.fs.Path;
  8. import org.apache.hadoop.io.LongWritable;
  9. import org.apache.hadoop.io.Text;
  10. import org.apache.hadoop.mapreduce.Job;
  11. import org.apache.hadoop.mapreduce.Mapper;
  12. import org.apache.hadoop.mapreduce.Reducer;
  13. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  14. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  16. public class PageView {
  18.     /**
  19.      * 利用mapreduce 离线统计每天的pv
  20.      * 
  21.      * @author itunic
  22.      *
  23.      */
  24.     public static class PageVisitsMapper extends Mapper<LongWritable, Text, KPI, LongWritable> {
  25.         @Override
  26.         protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
  27.             // 校验每一行URL的合法性
  28.             KPI kpi = KPI.filterPVs(value.toString());
  29.             if (kpi.isValid()) {
  30.                 // 利用mapper特性输出给reducer
  31.                 context.write(kpi, new LongWritable(1));
  32.             }
  33.         }
  34.     }
  36.     public static class PageVisitsReducer extends Reducer<KPI, LongWritable, Text, Text> {
  37.         @Override
  38.         protected void reduce(KPI key, Iterable<LongWritable> value, Context context)
  39.                 throws IOException, InterruptedException {
  40.             long count = 0;
  41.             // 将相同的分组相同的页面循环叠加
  42.             for (LongWritable l : value) {
  43.                 count += l.get();
  44.             }
  45.             String out = key.getRequest() + "\t" + count;
  46.             try {
  47.                 context.write(new Text(key.getTime_local_day()), new Text(out));
  48.             } catch (ParseException e) {
  49.                 // TODO Auto-generated catch block
  50.                 e.printStackTrace();
  51.             }
  52.         }
  53.     }
  55.     public static void main(String[] args) throws Exception {
  56.         Configuration conf = new Configuration();
  57.         Job job = Job.getInstance(conf);
  58.         job.setJarByClass(PageView.class);
  59.         job.setMapperClass(PageVisitsMapper.class);
  60.         job.setReducerClass(PageVisitsReducer.class);
  61.         job.setMapOutputKeyClass(KPI.class);
  62.         job.setMapOutputValueClass(LongWritable.class);
  63.         job.setOutputKeyClass(Text.class);
  64.         job.setOutputValueClass(Text.class);
  65.         job.setGroupingComparatorClass(PageViewGroup.class);
  66.         FileInputFormat.setInputPaths(job, new Path("F:\\test\\input\\access.log.fensi"));
  67.         FileOutputFormat.setOutputPath(job, new Path("F:\\test\\output9"));
  69.         int i = job.waitForCompletion(true) ? 0 : 1;
  70.         System.exit(i);
  71.     }
  72. }

源数据实例：nginx log

  1. 194.237.142.21 - - [18/Sep/2013:06:49:18 +0000] "GET /wp-content/uploads/2013/07/rstudio-git3.png HTTP/1.1" 304 0 "-" "Mozilla/4.0 (compatible;)"

 

运行分析部分结果如下：

  1. 2013-09-19  /tag/hadoop/page/3//images/stories/3xp.php  2
  2. 2013-09-19  /tag/rhadoop//images/stories/3xp.php    2
  3. 2013-09-19  /ucp.php    1
  4. 2013-09-19  /wp-admin/admin-ajax.php    120
  5. 2013-09-19  /wp-admin/async-upload.php  3
  6. 2013-09-19  /wp-admin/edit-comments.php 1
  7. 2013-09-19  /wp-admin/post.php  4
  8. 2013-09-19  /wp-cron.php    24
  9. 2013-09-19  /wp-login.php   2
  10. 2013-09-19  /xmlrpc.php 3
  11. 2013-09-18  //components/com_jnews/includes/openflashchart/php-ofc-library/ofc_upload_image.php 1
  12. 2013-09-18  //images/stories/3xp.php    6
  13. 2013-09-18  //images/stories/70cpx.php  2
  14. 2013-09-18  //images/stories/70pet.php  2
  15. 2013-09-18  //images/stories/cr0t.php   1
  16. 2013-09-18  /admin.php  2
  17. 2013-09-18  /administrator/index.php    2
  18. 2013-09-18  /batch.manage.php   21
  19. 2013-09-18  /category/hadoop-action/page/4//images/stories/3xp.php  2
  20. 2013-09-18  /images/stories/cr0t.php    1
  21. 2013-09-18  /index.php  10
  22. 2013-09-18  /index.php/blog/14  1
  23. 2013-09-18  /index.php/quiz/1   1
  24. 2013-09-18  /index.php/quiz/2   1

点击下载测试数据及分析结果数据

下载测试数据

相关

利用Apache Spark实现pv统计分析

文件下载

广告也精彩
测试数据
  • 利用Hadoop Mapreduce实现pv统计分析已关闭评论
  • 456 views
  • A+
所属分类：未分类
avatar