大数据学习day7------hadoop04----1 流量案例 2 电影案例(统计每部电影的均分，统计每个人的均分，统计电影的评论次数，***统计每部电影评分最高的N条记录(Integer.max)，统计评论次数最多的n部电影（全局排序）) 3 line线段重叠次数案例 4.索引案例

public class JobDriver {
    public static void main(String[] args) throws Exception {
         // 获取MR程序运行时的初始化配置
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);
        // 设置map和reduce类，调用类中自定义的map  reduce方法的业务逻辑
        job.setMapperClass(ViewsMapper.class);
        job.setReducerClass(ViewsReducer.class);
        // 设置map端输出key-value的类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(LongWritable.class);
        // 设置reduce端输出key-value的类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);
        // 处理的文件的路径
        FileInputFormat.setInputPaths(job, new Path("E:/wc/flow.txt"));
        // 结果输出路径
        FileOutputFormat.setOutputPath(job, new Path("E:/wc/flow/"));
        // 提交任务，参数   等待执行
        job.waitForCompletion(true);
    }
}

public class MovieAvgRate {
    static Text k = new Text();
    static DoubleWritable v = new DoubleWritable();
    // mapper部分
    static class MovieMapper extends Mapper<LongWritable, Text, Text, DoubleWritable>{
        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, DoubleWritable>.Context context)
                throws IOException, InterruptedException {
            String line = value.toString();
            MovieBean m = JSON.parseObject(line, MovieBean.class);
            String movie = m.getMovie();
            double rate = m.getRate();
            k.set(movie);
            v.set(rate);
            context.write(k, v);
        }
    }
    // Reducer部分
    static class MovieReducer extends Reducer<Text, DoubleWritable, Text, DoubleWritable>{
        
        @Override
        protected void reduce(Text key, Iterable<DoubleWritable> iters,
                Reducer<Text, DoubleWritable, Text, DoubleWritable>.Context context)
                throws IOException, InterruptedException {
            double sumRate = 0;
            int count = 0;
            for (DoubleWritable doubleWritable : iters) {
                double rate = doubleWritable.get();
                sumRate += rate;
                count++;
            }
            double avg = sumRate/count;
            v.set(avg);
            context.write(key, v);
        }
    }
    // JOB驱动
    public static void main(String[] args) throws Exception {
         // 获取MR程序运行时的初始化配置
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);
        // 设置map和reduce类，调用类中自定义的map  reduce方法的业务逻辑
        job.setMapperClass(MovieMapper.class);
        job.setReducerClass(MovieReducer.class);
        // 设置map端输出key-value的类型，当map和reduce的输出值类型一致时，可以省略map处值类型的设置
//        job.setMapOutputKeyClass(Text.class);
//        job.setMapOutputValueClass(LongWritable.class);
        // 设置reduce端输出key-value的类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(DoubleWritable.class);
        // 处理的文件的路径
        FileInputFormat.setInputPaths(job, new Path("E:/javafile/movie.txt"));
        // 结果输出路径
        FileOutputFormat.setOutputPath(job, new Path("E:/wc/movie/res2"));
        // 提交任务，参数   等待执行
        job.waitForCompletion(true);
    }
}