课堂测试-数据清洗2

题目：

Result文件数据说明：

Ip：106.39.41.166,（城市）

Date：10/Nov/2016:00:01:02 +0800,（日期）

Day：10,（天数）

Traffic: 54 ,（流量）

Type: video,（类型：视频video或文章article）

Id: 8701（视频或者文章的id）

测试要求：

1、 数据清洗：按照进行数据清洗，并将清洗后的数据导入hive数据库中。

两阶段数据清洗：

（1）第一阶段：把需要的信息从原始日志中提取出来

ip:    199.30.25.88

time:  10/Nov/2016:00:01:03 +0800

traffic:  62

文章： article/11325

视频： video/3235

1 2 4 5 6

（2）第二阶段：根据提取出来的信息做精细化操作

ip--->城市 city（IP）

date--> time:2016-11-10 00:01:03

day: 10

traffic:62

type:article/video

id:11325

（3）hive数据库表结构:

create table data(  ip string,  time string , day string, traffic bigint,

type string, id   string )

2、数据处理：

·统计最受欢迎的视频/文章的Top10访问次数 （video/article）

·按照地市统计最受欢迎的Top10课程 （ip）

·按照流量统计最受欢迎的Top10课程 （traffic）

3、数据可视化：将统计结果倒入MySql数据库中，通过图形化展示的方式展现出来。

完成情况:

目前完成了第一，二步。下面拿第二步的第一个来说：

第二步中主要是对数据进行处理，一开始看到这个题目，想到的是用mapreduce一步解决，但是不仅需要将相同id的video/article进行总和得到新的一列数据num，还要对num进行倒顺处理，一步解决难度有些大，所以进行两次mapreduce来对数据进行处理，然后导入hive，查询前十个数据即为所求。

第一个步骤：将相同id的video/article进行总和得到新的一列数据num，并且只保留num与id这两列数据。

import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
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;

public class text_2_1 {
    public static class Map extends Mapper<Object,Text,Text,Text>{
        private static Text newKey = new Text();
        private static Text newvalue = new Text("1");
        public void map(Object key,Text value,Context context) throws IOException, InterruptedException{
        String line = value.toString();
        String arr[] = line.split(" ");
        newKey.set(arr[5]);
        context.write(newKey,newvalue);
        }
    }
        public static class Reduce extends Reducer<Text, Text, Text, Text> {
            private static Text newkey = new Text();
            private static Text newvalue = new Text();
            protected void reduce(Text key, Iterable<Text> values, Context context)throws IOException, InterruptedException {
                int num = 0;
                    for(Text text : values){
                        num++;
                    }
                    newkey.set(""+num);
                    newvalue.set(key);
                    context.write(newkey,newvalue);
            }
        }
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
    Configuration conf = new Configuration();
    conf.set("mapred.textoutputformat.separator", " ");
    System.out.println("start");
      Job job=Job.getInstance(conf); 
      job.setJarByClass(text_2_1.class);
      job.setMapperClass(Map.class);
      job.setReducerClass(Reduce.class);
      job.setOutputKeyClass(Text.class); 
      job.setOutputValueClass(Text.class); 
      Path in=new Path("hdfs://localhost:9000/text/in/data");
      Path out=new Path("hdfs://localhost:9000/text/out1");
      FileInputFormat.addInputPath(job, in);
      FileOutputFormat.setOutputPath(job, out);
      boolean flag = job.waitForCompletion(true);
        System.out.println(flag);
        System.exit(flag? 0 : 1);
  }
}

第二个步骤，对num进行倒顺处理。因为在MapReduce中默认为正序排序，所以新定义了一个比较的类。

import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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;
public class m {
    public static class Map extends Mapper<Object,Text,IntWritable,Text>{
        private static IntWritable  newKey = new IntWritable ();
        private static Text  newvalue = new Text ();
        public void map(Object key,Text value,Context context) throws IOException, InterruptedException{
        String line = value.toString();
        String arr[] = line.split(" ");
        newKey.set(Integer.parseInt(arr[0]));
        newvalue.set(arr[1]);
        context.write(newKey,newvalue);
        }
    }
        public static class Reduce extends Reducer<IntWritable, Text, IntWritable, Text> {
            protected void reduce(IntWritable key, Iterable<Text> values, Context context)throws IOException, InterruptedException {
                    for(Text text : values){
                        context.write(key,text);
                    }
            }
        }
        public static class IntWritableDecreasingComparator extends    IntWritable.Comparator 
        {        
            public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) 
            {            
                return -super.compare(b1, s1, l1, b2, s2, l2);        
                }    
            }
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
    Configuration conf = new Configuration();
    conf.set("mapred.textoutputformat.separator", " ");
    System.out.println("start");
      Job job=Job.getInstance(conf); 
      job.setJarByClass(m.class);
      job.setMapperClass(Map.class);
      job.setReducerClass(Reduce.class);
      job.setOutputKeyClass(IntWritable.class); 
      job.setOutputValueClass(Text.class); 
      job.setSortComparatorClass(IntWritableDecreasingComparator.class);
      Path in=new Path("hdfs://localhost:9000/text/out1/part-r-00000");
      Path out=new Path("hdfs://localhost:9000/text/out2");
      FileInputFormat.addInputPath(job, in);
      FileOutputFormat.setOutputPath(job, out);
      boolean flag = job.waitForCompletion(true);
        System.out.println(flag);
        System.exit(flag? 0 : 1);
  }
}

经过了上面两步骤的处理后，得到的数据是这样的：

 然后将其导入hive数据库，再查询前十行，即为所求。