各类异常汇总
https://www.cnblogs.com/spicy/p/9645792.html
https://blog.csdn.net/diyangxia/article/details/78982867
https://www.cnblogs.com/LeachBlog/p/9968052.html

步骤

继承Mapper，编写map的业务逻辑
继承Reducer，编写reduce的业务逻辑
编写Driver驱动类

Mapper类

  1. package com.plat.mapreduce;
  2. import org.apache.hadoop.io.IntWritable;
  3. import org.apache.hadoop.io.Text;
  4. import org.apache.hadoop.mapreduce.Mapper;
  5. import java.io.IOException;
  7. /**
  8.  * Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>
  9.  *     KEYIN - 是指框架读取到的数据的key类型
  10.  *          在默认的读取数据组件InputFormat下，读取的key是一行文本的偏移量，所以key的类型是long类型的
  11.  *     VALUEIN - 是指框架读取到的数据的value类型
  12.  *          在默认的读取数据组件InputFormat下，读到的value就是一行文本的内容，所以value的类型是String类型的
  13.  *     KEYOUT - 是指用户自定义逻辑方法返回的数据中key的类型，这个是由用户业务逻辑决定的
  14.  *          在我们单词统计当中，我们输出的是单词数量作为value，所以类型是String
  15.  *     VALUEOUT - 是指用户自定义逻辑方法返回的数据中value的类型，这个是由用户逻辑决定的
  16.  *          在我们做单词统计当中，我们输出的单词数量作为value，所以这个类型是Integer
  17.  * String、long是jdk自带的数据类型，序列化效率低，因此hadoop为了提高序列化速度，自己定义了一套数据结构
  18.  *      String --- Text
  19.  *      Integer --- IntWritable
  20.  *      null --- nullWritable
  21.  */
  22. // 注意引入mapper的包
  23. public class WordcountMapper extends Mapper {
  25.     @Override
  26.     protected void map(Object key, Object value, Context context) throws IOException, InterruptedException {
  27.         //key为每行的起始偏移量
  28.         System.out.println("每行的key值 ============ " + key);
  29.         //value为每行的数据
  30.         System.out.println("每行的value值" + value.toString());
  31.         String line = value.toString();
  32.         String[] strArr = line.split(" ");
  33.         for (String str : strArr) {
  34.             context.write(new Text(str), new IntWritable(1));
  35.         }
  37.     }
  38. }

Reducer类

  1. package com.plat.mapreduce;
  3. import org.apache.hadoop.io.IntWritable;
  4. import org.apache.hadoop.io.Text;
  5. import org.apache.hadoop.mapreduce.Reducer;
  7. import java.io.IOException;
  9. /**
  10.  * reduceTask在调用我们的reduce方法
  11.  * reduceTask应该接收map阶段所有maptask输出的数据中的一部分
  12.  *      （key.hashcode%numReduceTask==本ReduceTask编号）
  13.  * 接收到的所有kv会按照key进行分组，然后将kv传给reduce
  14.  *
  15.  *
  16.  *
  17.  */
  18. public class WordcountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
  19.     // key为map阶段输出的key
  20.     // value为上一阶段相同key输出对应的value集合
  21.     @Override
  22.     protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
  23.         int count = 0;
  24.         for(IntWritable v :values){
  25.             count = count + v.get();
  26.         }
  27.         context.write(key, new IntWritable(count));
  28.     }
  29. }

编写Driver驱动类

  1. package com.plat.mapreduce;
  3. import org.apache.hadoop.conf.Configuration;
  4. import org.apache.hadoop.fs.Path;
  5. import org.apache.hadoop.io.IntWritable;
  6. import org.apache.hadoop.io.Text;
  7. import org.apache.hadoop.mapreduce.Job;
  8. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  9. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
  10. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  11. import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
  13. public class WordcountDriver {
  15.     public static void main(String[] args) throws Exception{
  16.         Configuration configuration = new Configuration();
  17.         //如果不做任何配置，默认采用本地文件系统及本地yarn
  18.         //采用指定的hdfs文件系统
  19.         configuration.set("fs.defaultFS", "hdfs://sunjz-01:9000");
  20.         //configuration.set("mapreduce.framework.name", "yarn");
  21.         //configuration.set("yarn.resourcemanager.hostname", "sunjz-02");
  22.         //configuration.set("mapreduce.app-submission.cross-platform", "true");
  23.         Job job = Job.getInstance(configuration);
  25.         //告诉程序jar包所在的位置
  26.         //job.setJar("/root/wordcount.jar"); //手动写死
  27.         job.setJarByClass(WordcountDriver.class); //自动扫描
  28.         //告诉框架我们所需的map类和reduce类
  29.         job.setMapperClass(WordcountMapper.class);
  30.         job.setReducerClass(WordcountReducer.class);
  31.         //告诉框架我们程序输出的数据类型
  32.         job.setMapOutputKeyClass(Text.class);
  33.         job.setMapOutputValueClass(IntWritable.class);
  35.         job.setOutputKeyClass(Text.class);
  36.         job.setOutputValueClass(IntWritable.class);
  38.         //告诉程序我们使用的数据读取组件，结果输出组件是什么
  39.         job.setInputFormatClass(TextInputFormat.class);
  40.         job.setOutputFormatClass(TextOutputFormat.class);
  41.         //告诉框架我们要处理的数据文件放在哪个路径
  42.         //FileInputFormat.setInputPaths(job, new Path("D:\\wordcount\\input"));
  43.         //FileOutputFormat.setOutputPath(job, new Path("D:\\wordcount\\output"));
  45.         FileInputFormat.setInputPaths(job, new Path("/wordcount/input"));
  46.         FileOutputFormat.setOutputPath(job, new Path("/wordcount/output"));
  48.         boolean flag = job.waitForCompletion(true);
  49.         System.exit(flag ? 0 : 1);
  50.     }
  51. }