海量Web日志分析（MR实现）

代码需求笔者配置数据类型需求具体实现思路代码结构

KPI.javaKPIPV.java

代码结果截图 KPIIP.java

代码结果截图 KPITime.java

代码结果截图（部分） KPIBrowser.java

代码结果截图（部分）

海量Web日志分析（MR实现）代码需求

实现PV，IP，访问时间，浏览器类型等常用的Web日志分析功能。

笔者配置

Hadoop3.1.3，有空再用spark实现
IDEA本地测试运行，可以打jar包上传集群跑数据

数据类型

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

146190行数据，其中包含脏数据，以空格为分割符

需求具体实现思路

KPI.java: 对数据的处理，还有一些功能类KPIPV.java: 统计特定页面的访问数量KPIIP.java: 统计特定页面的访问独立ip数量（存在相同IP访问多次）KPITime.java：每个时间段的访问数量KPIBrowser.java：统计每种浏览器的访问数量代码结构

KPI.java

package com.lmr;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashSet;
import java.util.Locale;
import java.util.Set;

public class KPI {
    private String remote_addr;     // 记录客户端的ip地址
    private String remote_user;     // 记录客户端用户名称,忽略属性"-"
    private String time_local;      // 记录访问时间与时区
    private String request;         // 记录请求的url与http协议
    private String status;          // 记录请求状态；成功是200
    private String body_bytes_sent; // 记录发送给客户端文件主体内容大小
    private String http_referer;    // 用来记录从哪个页面链接访问过来的
    private String http_user_agent; // 记录客户浏览器的相关信息

    private boolean valid = true;   // 判断数据是否合法

    private static KPI parser(String line){
        KPI kpi = new KPI();
        String[] arr = line.split(" ");
        if (arr.length > 11){
            kpi.setRemote_addr(arr[0]);
            kpi.setRemote_user(arr[1]);
            kpi.setTime_local(arr[3].substring(1));
            kpi.setRequest(arr[6]);
            kpi.setStatus(arr[8]);
            kpi.setBody_bytes_sent(arr[9]);
            kpi.setHttp_referer(arr[10]);

            if (arr.length > 12){
                kpi.setHttp_user_agent(arr[11] + " " + arr[12]);
            }else {
                kpi.setHttp_user_agent(arr[11]);
            }

            if (Integer.parseInt(kpi.getStatus()) >= 400){
                kpi.setValid(false);
            }
        }else {
            kpi.setValid(false);
        }
        return kpi;
    }


    
    public static KPI filterPVs(String line){
        KPI kpi = parser(line);
        Set pages = new HashSet();

        pages.add("/about");
        pages.add("/black-ip-list/");
        pages.add("/cassandra-clustor/");
        pages.add("/finance-rhive-repurchase/");
        pages.add("/hadoop-family-roadmap/");
        pages.add("/hadoop-hive-intro/");
        pages.add("/hadoop-zookeeper-intro/");
        pages.add("/hadoop-mahout-roadmap/");

        if (!pages.contains(kpi.getRequest())){
            kpi.setValid(false);
        }
        return kpi;
    }


    
    public static KPI filterIPs(String line){
        KPI kpi = parser(line);

        Set pages = new HashSet<>();

        pages.add("/about");
        pages.add("/black-ip-list/");
        pages.add("/cassandra-clustor/");
        pages.add("/finance-rhive-repurchase/");
        pages.add("/hadoop-family-roadmap/");
        pages.add("/hadoop-hive-intro/");
        pages.add("/hadoop-zookeeper-intro/");
        pages.add("/hadoop-mahout-roadmap/");

        if (!pages.contains(kpi.getRequest())){
            kpi.setValid(false);
        }
        return kpi;
    }

    
    public static KPI filterBroswer(String line){
        return parser(line);
    }


    
    public static KPI filterTime(String line){
        return parser(line);
    }

    
    public static KPI filerDomain(String line){
        return parser(line);
    }

    
    public String toString(){
        StringBuilder sb = new StringBuilder();
        sb.append("valid:" + this.valid);
        sb.append("--remote_addr:" + this.remote_addr);
        sb.append("--remote_user:" + this.remote_user);
        sb.append("--time_local:" + this.time_local);
        sb.append("--request:" + this.request);
        sb.append("--status:" + this.status);
        sb.append("--body_bytes_sent:" + this.body_bytes_sent);
        sb.append("--http_referer:" + this.http_referer);
        sb.append("--http_user_agent:" + this.http_user_agent);
        return sb.toString();
    }


    public Date getTime_local_Date() throws ParseException {
        SimpleDateFormat df = new SimpleDateFormat("dd/MMM/yyyy:HH:mm:ss", Locale.US);
        return df.parse(this.time_local);
    }
    
    public String getTime_local_Date_hour() throws ParseException {
        SimpleDateFormat df = new SimpleDateFormat("yyyyMMddHH");
        return df.format(this.getTime_local_Date());
    }

    public String getHttp_referer_domain() {
        if (http_referer.length() < 8) {
            return http_referer;
        }

        String str = this.http_referer.replace(""", "").replace("http://", "").replace("https://", "");
        return str.indexOf("/") > 0 ? str.substring(0, str.indexOf("/")) : str;
    }
    

    public String getRemote_addr() {
        return remote_addr;
    }

    public void setRemote_addr(String remote_addr) {
        this.remote_addr = remote_addr;
    }

    public String getRemote_user() {
        return remote_user;
    }

    public void setRemote_user(String remote_user) {
        this.remote_user = remote_user;
    }

    public String getTime_local() {
        return time_local;
    }

    public void setTime_local(String time_local) {
        this.time_local = time_local;
    }

    public String getRequest() {
        return request;
    }

    public void setRequest(String request) {
        this.request = request;
    }

    public String getStatus() {
        return status;
    }

    public void setStatus(String status) {
        this.status = status;
    }

    public String getBody_bytes_sent() {
        return body_bytes_sent;
    }

    public void setBody_bytes_sent(String body_bytes_sent) {
        this.body_bytes_sent = body_bytes_sent;
    }

    public String getHttp_referer() {
        return http_referer;
    }

    public void setHttp_referer(String http_referer) {
        this.http_referer = http_referer;
    }

    public String getHttp_user_agent() {
        return http_user_agent;
    }

    public void setHttp_user_agent(String http_user_agent) {
        this.http_user_agent = http_user_agent;
    }

    public boolean isValid() {
        return valid;
    }

    public void setValid(boolean valid) {
        this.valid = valid;
    }


    
    public static void main(String[] args) {
        String line = "222.68.172.190 - - [18/Sep/2013:06:49:57 +0000] "GET /images/my.jpg HTTP/1.1" 200 19939 "http://www.chinasofti.com" "Mozilla/5.0 (Windows NT 6.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/51.0.1547.66 Safari/537.36"";

        System.out.println(line);

        KPI kpi = parser(line);
        System.out.println(kpi);

        try {
            SimpleDateFormat df = new SimpleDateFormat("yyyy.MM.dd:HH:mm:ss", Locale.US);
            System.out.println(df.format(kpi.getTime_local_Date()));
            System.out.println(kpi.getTime_local_Date_hour());
            System.out.println(kpi.getHttp_referer_domain());
        } catch (ParseException e) {
            e.printStackTrace();
        }
    }
    
}

KPIPV.java

package com.lmr;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;


public class KPIPV {
    public static class KPIPVMapper extends Mapper{
        private Text outK = new Text();
        private IntWritable outV = new IntWritable(1);

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            KPI kpi = KPI.filterPVs(value.toString());
            if (kpi.isValid()){
                outK.set(kpi.getRequest());
                System.out.println(kpi);
                context.write(outK, outV);
            }
        }
    }

    public static class KPIPVReducer extends Reducer{
        private IntWritable outV = new IntWritable();

        @Override
        protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {
            int sum = 0;

            for (IntWritable value : values) {
                sum += value.get();
            }
            outV.set(sum);

            context.write(key, outV);
        }
    }

    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {

        String input = "海量Web日志分析系统/src/datas/input";
        String output = "海量Web日志分析系统/src/datas/output/pv1";

        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "KPIPV");
        job.setJarByClass(KPIPV.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        job.setMapperClass(KPIPVMapper.class);
        job.setReducerClass(KPIPVReducer.class);

        FileInputFormat.addInputPath(job, new Path(input));
        FileOutputFormat.setOutputPath(job, new Path(output));


        if (!job.waitForCompletion(true)){
            return;
        }
    }

}

代码结果截图

KPIIP.java

package com.lmr;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.util.HashSet;
import java.util.Set;

public class KPIIP {
    public static class KPIIPMapper extends Mapper {
        private Text outK = new Text();
        private Text outV = new Text();

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            KPI kpi = KPI.filterIPs(value.toString());
            if (kpi.isValid()){
                outK.set(kpi.getRequest());
                outV.set(kpi.getRemote_addr());
                context.write(outK, outV);
            }
        }
    }

    public static class KPIIPReducer extends Reducer {
        private Text outV = new Text();

        @Override
        protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {

            // 放在set里面，集合元素不会重复
            Set count = new HashSet();

            for (Text value : values) {
                count.add(value.toString());
            }
            outV.set(String.valueOf(count.size()));
            context.write(key, outV);
        }
    }

    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {

        String input = "海量Web日志分析系统/src/datas/input";
        String output = "海量Web日志分析系统/src/datas/output/ip";

        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "KPIip");
        job.setJarByClass(KPIIP.class);

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

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

        job.setMapperClass(KPIIPMapper.class);
        job.setReducerClass(KPIIPReducer.class);

        FileInputFormat.addInputPath(job, new Path(input));
        FileOutputFormat.setOutputPath(job, new Path(output));

        if (!job.waitForCompletion(true)){
            return;
        }
    }
}

代码结果截图

KPITime.java

package com.lmr;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.text.ParseException;

public class KPITime {
    public static class KPITimeMapper extends Mapper{
        private Text outK = new Text();
        private IntWritable outV = new IntWritable(1);

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            KPI kpi = KPI.filterTime(value.toString());
            if (kpi.isValid()){
                try {
                    outK.set(kpi.getTime_local_Date_hour());
                    context.write(outK, outV);
                } catch (ParseException e) {
                    e.printStackTrace();
                }

            }
        }
    }

    public static class KPITimeReducer extends Reducer{
        private IntWritable outV = new IntWritable();

        @Override
        protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable value : values) {
                sum += value.get();
            }
            outV.set(sum);
            context.write(key, outV);
        }
    }

    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {

        String input = "海量Web日志分析系统/src/datas/input";
        String output = "海量Web日志分析系统/src/datas/output/time";

        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "KPITime");
        job.setJarByClass(KPITime.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        job.setMapperClass(KPITimeMapper.class);
        job.setReducerClass(KPITimeReducer.class);

        FileInputFormat.addInputPath(job, new Path(input));
        FileOutputFormat.setOutputPath(job, new Path(output));

        if (!job.waitForCompletion(true)){
            return;
        }
    }
}

代码结果截图（部分）

KPIBrowser.java

package com.lmr;

import com.sun.xml.internal.txw2.TxwException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class KPIBrowser {
    public static class KPIBrowserMapper extends Mapper{
        private Text outK = new Text();
        private IntWritable outV = new IntWritable(1);

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            KPI kpi = KPI.filterBroswer(value.toString());
            if (kpi.isValid()){
                outK.set(kpi.getHttp_user_agent());
//                System.out.println(kpi.getHttp_user_agent());
                context.write(outK, outV);
            }
        }
    }

    public static class KPIBrowserReducer extends Reducer{
        private IntWritable outV = new IntWritable();

        @Override
        protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable value : values) {
                sum += value.get();
            }
            outV.set(sum);
            context.write(key, outV);
        }
    }

    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {

        String input = "海量Web日志分析系统/src/datas/input";
        String output = "海量Web日志分析系统/src/datas/output/browser1";

        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "KPIBrowser");
        job.setJarByClass(KPIBrowser.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        job.setMapperClass(KPIBrowserMapper.class);
        job.setReducerClass(KPIBrowserReducer.class);

        FileInputFormat.addInputPath(job, new Path(input));
        FileOutputFormat.setOutputPath(job, new Path(output));

        if (!job.waitForCompletion(true)){
            return;
        }
    }
}

代码结果截图（部分）

海量Web日志分析（MR实现）

大数据系统相关栏目本月热门文章