package cn.gigdata.hdfs.mr;

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Mapper;

/**

 * 参数信息：

 *     1.KEYIN: 默认情况下，是mr框架所读到的一行文本的起始偏移量，Long,在hadoop中用自己的序列化接口，不直接用Long，用LongWritable

 *  　 2.VALUEIN:默认情况下，是mr框架所读到的一行文本的内容，String，同上，用String的序列化接口Text

 *     3.KEYOUT：是用户自定义逻辑处理完成之后输出数据中的key，在此处是单词，String，同上，用Text

 *     4.VALUEOUT：是用户自定义逻辑处理完成之后输出数据中的value，在此处是单词次数，Integer，同上，用IntWritable

 *  Map程序只写一份，但是可以被很多Map Task调用

 */

public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{

    /**

     * map阶段的业务逻辑就写在自定义的map()方法中

     * maptask会对每一行输入数据调用一次我们自定义的map()方法

     */

    @Override

    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

        //将maptask传给我们的文本内容先转换成String

        String line = value.toString();

        //根据空格将这一行切分成单词

        String[] words = line.split(" ");

        //将单词输出为<单词，1>

        for(String word:words){

            //将单词作为key，将次数1作为value，以便于后续根据单词进行数据分发，以便于相同单词会到相同的reduce task

            context.write(new Text(word), new IntWritable());

        }

    }

}   

package cn.gigdata.hdfs.mr;

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

/**

 * KEYIN, VALUEIN 对应  mapper输出的KEYOUT,VALUEOUT类型对应

 * KEYOUT, VALUEOUT 是自定义reduce逻辑处理结果的输出数据类型： KEYOUT是单词   VLAUEOUT是总次数

 */

public class WordcountReducer extends Reducer<Text, IntWritable, Text, IntWritable>{

    /**

     * <angelababy,1><angelababy,1><angelababy,1><angelababy,1>

     * <hello,1><hello,1><hello,1><hello,1><hello,1><hello,1>

     * <banana,1><banana,1><banana,1><banana,1><banana,1><banana,1>

     * 入参key，是一组相同单词kv对应的key（以组为单位调一次）reduce传key的时候传的是组中的第一个

     */

    //生命周期：框架每传递进来一个kv 组，reduce方法被调用一次

    @Override

    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {

        int count=0;

        /*Iterator<IntWritable> iterator = values.iterator();

        while(iterator.hasNext()){

            count += iterator.next().get();

        }*/

        for(IntWritable value:values){

            count += value.get();

        }

        context.write(key, new IntWritable(count));//默认将结果写入（HDFS）文件

    }

}

package cn.gigdata.hdfs.mr;

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.lib.input.FileInputFormat;

import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**

 * 相当于一个yarn集群的客户端

 * 需要在此封装我们的mr程序的相关运行参数，指定jar包

 * 最后提交给yarn

 * @author

 *

 */

//定义一个主类，用来描述job并提交job

public class WordcountDriver {

    //把业务逻辑相关的信息（哪个是mapper，哪个是reducer，要处理的数据在哪里，输出的结果放哪里……）描述成一个job对象

    //把这个描述好的job提交给集群去运行

    public static void main(String[] args) throws Exception {

        if (args == null || args.length == 0) {

            args = new String[2];

            args[0] = "hdfs://shizhan2:9000/wordcount/input/wordcount.txt";

            args[1] = "hdfs://shizhan2:9000/wordcount/output8";

        }

        Configuration conf = new Configuration();

        Job job = Job.getInstance(conf);

        //指定本程序的jar包所在的本地路径

        job.setJarByClass(WordcountDriver.class);

        //指定本业务job要使用的mapper/Reducer业务类

        job.setMapperClass(WordCountMapper.class);

        job.setReducerClass(WordcountReducer.class);

        //指定mapper输出数据的kv类型

        job.setMapOutputKeyClass(Text.class);

        job.setMapOutputValueClass(IntWritable.class);

        //指定最终输出的数据的kv类型

        job.setOutputKeyClass(Text.class);

        job.setOutputValueClass(IntWritable.class);

        //指定job的输入原始文件所在目录

        FileInputFormat.setInputPaths(job, new Path(args[0]));

        //指定job的输出结果所在目录

        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        //将job中配置的相关参数，以及job所用的java类所在的jar包，提交给YARN集群中去运行

        /*job.submit();*/

        boolean res = job.waitForCompletion(true);

        System.exit(res?0:1);

    }

}