/*
 * 1.Mapper 类的四个泛型中，前两个指定 mapper 输入数据的类型,后两个指定 mapper 输出数据的类型
 *  KEYIN 是输入的 key 的类型，VALUEIN 是输入的 value 的类型
 *  KEYOUT 是输出的 key 的类型，VALUEOUT 是输出的 value 的类型
 * 2.map 和 reduce 的数据的输入输出都是以 key-value 对的形式封装的
 * 3.默认情况下，框架传递给我们的 mapper 的输入数据中，key 是要处理的文本中一行的起始偏移量，为 Long 类型，
 * 这一行的内容为 value，为 String 类型的 
 * 4.后两个泛型的赋值需要我们结合实际情况
 * 5.为了在网络中传输时序列化更高效，Hadoop 把 Java 中的 Long 封装为 LongWritable, 把 String 封装为 Text
 */
public class WCMapper extends Mapper<LongWritable, Text, Text, LongWritable> {
  
  //重写 Mapper 中的 map 方法，MapReduce 框架每读一行数据就调用一次此方法
  @Override
  protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

    //书写具体的业务逻辑，业务要处理的数据已经被框架传递进来，就是方法的参数中的 key 和 value
    //key 是这一行数据的起始偏移量，value 是这一行的文本内容
    
    //1.将 Text 类型的一行的内容转为 String 类型
    String line = value.toString();
    
    //2.使用 StringUtils 以空格切分字符串，返回 String[]
    String[] words = StringUtils.split(line, " ");
    
    //3.循环遍历 String[],调用 context 的 writer()方法，输出为 key-value 对的形式
    //key：单词  value：1
    for(String word : words) {
      context.write(new Text(word), new LongWritable(1));
    }
    
  }

}

/*
 * 1.Reducer 类的四个泛型中，前两个输入要与 Mapper 的输出相对应。输出需要联系具体情况自定义
 */
public class WCReducer extends Reducer<Text, LongWritable, Text, LongWritable> {
  
  //框架在 map 处理完之后，将所有的 kv 对缓存起来，进行分组，然后传递一个分组(<key,{values}>,例如：<"hello",{1,1,1,1}>),
  //调用此方法
  @Override
  protected void reduce(Text key, Iterable<LongWritable> values, Context context)throws IOException, InterruptedException {
    
    //1.定义一个计数器
    long count = 0;
    
    //2.遍历 values的 list，进行累加求和
    for(LongWritable value : values) {
      //使用 LongWritable 的 get() 方法，可以将 一个 LongWritable 类型转为 Long 类型
      count += value.get();
    }
    
    //3.输出这一个单词的统计结果
    context.write(key, new LongWritable(count));
  }

}

/**
 * 此类用来描述一个特定的作业
 * 例：1.该作业使用哪个类作为逻辑处理中的 map，哪个作为 reduce
 *    2.指定该作业要处理的数据所在的路径
 *    3.指定该作业输出的结果放到哪个路径
 */
public class WCRunner {
  
  public static void main(String[] args) throws Exception {
    
    //1.获取 Job 对象:使用 Job 静态的 getInstance() 方法，传入 Configuration 对象
    Configuration conf = new Configuration();
    Job wcJob = Job.getInstance(conf);
    
    //2.设置整个 Job 所用的类的 jar 包：使用 Job 的 setJarByClass(),一般传入 当前类.class
    wcJob.setJarByClass(WCRunner.class);
    
    //3.设置本 Job 使用的 mapper 和 reducer 的类
    wcJob.setMapperClass(WCMapper.class);
    wcJob.setReducerClass(WCReducer.class);
    
    
    //4.指定 reducer 输出数据的 kv 类型 注：若 mapper 和 reducer 的输出数据的 kv 类型一致，可以用如下两行代码设置
    wcJob.setOutputKeyClass(Text.class);
    wcJob.setOutputValueClass(LongWritable.class);
    
    //5.指定 mapper 输出数据的 kv 类型
    wcJob.setMapOutputKeyClass(Text.class);
    wcJob.setMapOutputValueClass(LongWritable.class);
    
    //6.指定原始的输入数据存放路径:使用 FileInputFormat 的 setInputPaths() 方法
    FileInputFormat.setInputPaths(wcJob, new Path("/wc/srcdata/"));
    
    //7.指定处理结果的存放路径：使用 FileOutputFormat 的 setOutputFormat() 方法
    FileOutputFormat.setOutputPath(wcJob, new Path("/wc/output/"));
    
    //8.将 Job 提交给集群运行，参数为 true 表示显示运行状态
    wcJob.waitForCompletion(true);
    
  }

}