import java.io.IOException;


import org.apache.hadoop.conf.configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
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;



    public class Demo1 {

        // map类
        // 第一对kv,是决定数据输入的格式
        // 第二队kv 是决定数据输出的格式
        public static class map extends Mapper<LongWritable,Text,Text,LongWritable>{
            @override
            protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
                //一行一行的读，先LongWrtable再value是因为第一个LongWritable是偏移量
                String line=value.toString();
                //需要读出内容和行数1，所以要对结果进行类型转换
                context.write(new Text(line),new LongWritable(1));
            }
        }

        // reduce类
        // 用来接收map端输出的数据
        public static class reduce extends Reducer<Text,LongWritable,Text,LongWritable>{
            /**
             * reduce 聚合程序 每一个k都会调用一次
             * 默认是一个节点
             * key:每一个单词
             * values:map端 当前k所对应的所有的v
             */
            @Override
            protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {
                long sum=0;
                
                //进行整合后values值变成（key，1,1,1,1），values需要遍历
                for (LongWritable value : values) {
                    //这里同理需要将value转换类型，LongWritable是一个接口可以用get方法转为long型整数
                    sum+=value.get();
                }
                //同理long类型sum转换为LongWritable类型
                context.write(key,new LongWritable(sum));
            }
        }


        //mapreduce的程序入口
        public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
            //创造一个job任务
            Job job = Job.getInstance();
            //命名job名称
            job.setJobName("第一次通过自己的jar包连接");

            //指定当前main坐在类端口
            job.setJarByClass(Demo1.class);

            //指定map类端口
            job.setMapperClass(map.class);
            //指定map输出的kv类型
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(LongWritable.class);

            //指定reduce类端口
            job.setReducerClass(reduce.class);
            //指定reduce输出的kv类型
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(LongWritable.class);

            //指定输入路径 hDFs路径
            Path in = new Path("/wordcount");
            FileInputFormat.addInputPath(job,in);

            //指定输出路径
            Path out = new Path("/output1");
            //如果路径存在，进行删除操作
            FileSystem fs = FileSystem.get(new Configuration());
            if (fs.exists(out)){
                fs.delete(out,true);  //true可以删除多级目录
            }
            FileOutputFormat.setOutputPath(job,out);

            //启动任务
            job.waitForCompletion(true);

            /**
             * 提交任务
             * 1.通过maven中package将项目打包上传服务器然后执行
             * 2.执行任务 hadoop jar hadoop-mapreduce-examples-2.7.6.jar com.shujia.hadoop.Demo01WordCount /word  /output
             *
             */

            System.out.println("wordcount实现成功");


        }
    }

public static class map extends Mapper<LongWritable,Text,Text,LongWritable>{
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String s = value.toString();
            String[] split = s.split(",");
            for (String s1 : split) {
                context.write(new Text(s1),new LongWritable(1));
            }

        }
    }

public static class map extends Mapper<LongWritable,Text,Text,LongWritable>{
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String s = value.toString();
            String[] split = s.split(",");

            String s1 = split[2];
            LongWritable age = new LongWritable(Integer.valueOf(s1));
            String s2 = split[4];
            Text clazz = new Text(s2);
            context.write(clazz, age);
        }
    }