036 关于网站的UV分析

一:准备

1.统计的维度

　　guid

　　tracktime

　　provice

2.key与value的设定

　　key：date+provice_guid

　　value：NullWritable

3.案例分析

　　表示某天某个省份的某个人无论访问网站多少次，仅仅记做一次访问统计

　　UV：统计页面访问的总人数---》userID对于用户进行去重

二：程序

1.map程序

　　

 　　

2.reduce程序

　　

3.结果

　　

4.理解点

　　1）怎么去重

　　　　数据key的形式：date+provice_guid。

　　　　当guid是相同的时候，在shuffle的group分组时，key被分组，一起的放在一起，而value则是nullwritable，没有使用value。

　　　　所以到达reduce的时候，数据已经被去重了。

　　2）NullWritable.get()

　　　　使用反射，获得NullWritable的对象。

5.完整程序

package com.senior.network;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Mapper.Context;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class WebUvCount extends Configured implements Tool{
    //Mapper
    public static class WebUvCountMapper extends Mapper<LongWritable,Text,Text,NullWritable>{
        private Text mapoutputkey=new Text();
        @Override
        protected void cleanup(Context context) throws IOException,InterruptedException {
            
        }
        @Override
        protected void setup(Context context) throws IOException,InterruptedException {
            
        }
        
        @Override
        protected void map(LongWritable key, Text value, Context context)throws IOException, InterruptedException {
            String lineValue=value.toString();
            String[] strs=lineValue.split("	");
            if(30>strs.length){
                context.getCounter("webPvMapper_counter", "length_LT_30").increment(1L);
                return;
            }
            String guidValue=strs[5];    //
            if(StringUtils.isEmpty(guidValue)){
                return;
            }
            String trackTimeValue=strs[17];
            if(StringUtils.isEmpty(trackTimeValue)){
                return;
            }
            String dateVAlue=trackTimeValue.substring(0,13);//
            String priviceIdValue=strs[23];      
            
            Integer priviceId=Integer.MAX_VALUE;
            try{
                priviceId = Integer.valueOf(priviceIdValue);  //
            }catch(Exception e){
                return;
            }
            
            mapoutputkey.set(dateVAlue+"	"+priviceIdValue+"_"+guidValue);
            context.write(mapoutputkey,NullWritable.get());
        }
        
    }
    
    
    
    //Reducer
    public static class WebUvCountReducer extends Reducer<Text,NullWritable,Text,IntWritable>{
        private Text outputkey=new Text();
        private Map<String,Integer> dateMap;
        private IntWritable outputvalue=new IntWritable();
        
        @Override
        protected void setup(Context context)throws IOException, InterruptedException {
            dateMap=new HashMap<String,Integer>();
        }

        @Override
        protected void reduce(Text key, Iterable<NullWritable> values,Context context)throws IOException, InterruptedException {
            String date=key.toString().split("_")[0];
            if(dateMap.containsKey(date)){
                Integer previousUV=dateMap.get(date);
                Integer uv=previousUV+1;
                dateMap.put(date, uv);
            }else{
                dateMap.put(date, 1);
            }
        }

        @Override
        protected void cleanup(Context context)throws IOException, InterruptedException {
            Set<String> dateSet=dateMap.keySet();
            for(String date:dateSet){
                Integer uv=dateMap.get(date);
                outputkey.set(date);
                outputvalue.set(uv);
                context.write(outputkey, outputvalue);
            }
        }
        
        
    }
    
    //Driver
    public int run(String[] args)throws Exception{
        Configuration conf=this.getConf();
        Job job=Job.getInstance(conf,this.getClass().getSimpleName());
        job.setJarByClass(WebUvCount.class);
        //input
        Path inpath=new Path(args[0]);
        FileInputFormat.addInputPath(job, inpath);
        
        //output
        Path outpath=new Path(args[1]);
        FileOutputFormat.setOutputPath(job, outpath);
        
        //map
        job.setMapperClass(WebUvCountMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(NullWritable.class);
        
        //shuffle
        
        //reduce
        job.setReducerClass(WebUvCountReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        
        //submit
        boolean isSucess=job.waitForCompletion(true);
        return isSucess?0:1;
    }
    
    //main
    public static void main(String[] args)throws Exception{
        Configuration conf=new Configuration();
        //compress
        conf.set("mapreduce.map.output.compress", "true");
        conf.set("mapreduce.map.output.compress.codec", "org.apache.hadoop.io.compress.SnappyCodec");
        args=new String[]{
                "hdfs://linux-hadoop01.ibeifeng.com:8020/user/beifeng/mapreduce/wordcount/inputWebData",
                "hdfs://linux-hadoop01.ibeifeng.com:8020/user/beifeng/mapreduce/wordcount/outputWebData6"
        };
        int status=ToolRunner.run(new WebUvCount(), args);
        System.exit(status);
    }

}