大数据清洗阶段1

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

（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的基础学习，只实现了第一阶段里面数据的清洗 因为hive一直出错 没有实现把数据加载到hive里

这是wordcount代码 实现了对数据的统计个数 目前仅做到这儿了

今天不能及时完成原因：1.对MapReduce没有提前去学习 ，现在已经学了MapReduce一部分，明天计划把上次11个实验弄懂学会，并完成第二阶段以及排序。

2.没有提前对自己的hive进行测试，结果课上发现hive配置有错误。

package QingXi;
import java.io.IOException;
import java.util.StringTokenizer;
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 WordCount{
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
      Job job = Job.getInstance();
      job.setJobName("WordCount");
      job.setJarByClass(WordCount.class);
      job.setMapperClass(doMapper.class);
      job.setReducerClass(doReducer.class);
      job.setOutputKeyClass(Text.class);
      job.setOutputValueClass(IntWritable.class);
      Path in = new Path("hdfs://localhost:9000/user/hadoop/name/result.txt");  
      Path out = new Path("hdfs://localhost:9000/user/hadoop/name/out2");  
      FileInputFormat.addInputPath(job, in);  
      FileOutputFormat.setOutputPath(job, out);  
      System.exit(job.waitForCompletion(true) ? 0 : 1);  
    }  
    public static class doMapper extends Mapper<Object, Text, Text, IntWritable>{  
      public static final IntWritable one = new IntWritable(1);  
      public static Text word = new Text();  
      @Override  
       protected void map(Object key, Text value, Context context)  
               throws IOException, InterruptedException {  
          StringTokenizer tokenizer = new StringTokenizer(value.toString(), "");  
           word.set(tokenizer.nextToken());  
           context.write(word, one);  
      }  
    }  
    public static class doReducer extends Reducer<Text, IntWritable, Text, IntWritable>{  
      private IntWritable result = new IntWritable();  
         @Override
      protected void reduce(Text key, Iterable<IntWritable> values, Context context)  
      throws IOException, InterruptedException {  
      int sum = 0;  
      for (IntWritable value : values) {  
      sum += value.get();  
      }  
      result.set(sum);  
      context.write(key, result);  
      }  
    }  
}