mapreduce编程

跑一个mapreduce程序 需要yarn

执行命令：hadoop jar jar包文件路径（绝对的、相对的） 主函数 参数

  1. # 上传文件到 hadoop
  2. [root@hadoop01 hadoop3.2.1]# history |grep hadoop 
  3.   136  hadoop fs -mkdir /inhadoop fs -mkdir /in
  4.   137  hadoop fs -out LICENSE.txt /in/01.txt
  5.   138  hadoop fs -put LICENSE.txt /in/01.txt
  6.   139  hadoop fs -put LICENSE.txt /in/02.txt
  7.   140  hadoop fs -put LICENSE.txt /in/03.txt
  8. [root@hadoop01 mapreduce]# pwd
  9. /usr/local/hadoop3.2.1/share/hadoop/mapreduce
  10. [root@hadoop01 mapreduce]# hadoop jar hadoop-mapreduce-examples-3.2.1.jar wordcount /in /wcout

出现错误

  1. [2019-11-05 20:54:32.193]Container exited with a non-zero exit code 127. Error file: prelaunch.err.
  2. Last 4096 bytes of prelaunch.err :
  3. Last 4096 bytes of stderr :
  4. /bin/bash: /bin/java: No such file or directory
  7. [2019-11-05 20:54:32.193]Container exited with a non-zero exit code 127. Error file: prelaunch.err.
  8. Last 4096 bytes of prelaunch.err :
  9. Last 4096 bytes of stderr :
  10. /bin/bash: /bin/java: No such file or directory
  12. 添加环境变量都试过了，没有办法
  13. 算了，没有就给你一个吧（每个节点都给）
  14. ln -s /usr/local/java1.8/bin/java /bin/java

Mapreduce编程 - 图1

wordcount运行日志

  1. [root@hadoop01 mapreduce]# hadoop jar hadoop-mapreduce-examples-3.2.1.jar wordcount /in /wcout
  2. 2019-11-05 20:57:27,211 INFO client.RMProxy: Connecting to ResourceManager at hadoop02/192.168.61.102:8032
  3. 2019-11-05 20:57:28,037 INFO mapreduce.JobResourceUploader: Disabling Erasure Coding for path: /tmp/hadoop-yarn/staging/root/.staging/job_1572958454724_0004
  4. 2019-11-05 20:57:28,194 INFO sasl.SaslDataTransferClient: SASL encryption trust check: localHostTrusted = false, remoteHostTrusted = false
  5. 2019-11-05 20:57:28,436 INFO input.FileInputFormat: Total input files to process : 3
  6. 2019-11-05 20:57:28,500 INFO sasl.SaslDataTransferClient: SASL encryption trust check: localHostTrusted = false, remoteHostTrusted = false
  7. 2019-11-05 20:57:28,548 INFO sasl.SaslDataTransferClient: SASL encryption trust check: localHostTrusted = false, remoteHostTrusted = false
  8. 2019-11-05 20:57:28,971 INFO mapreduce.JobSubmitter: number of splits:3
  9. 2019-11-05 20:57:29,217 INFO sasl.SaslDataTransferClient: SASL encryption trust check: localHostTrusted = false, remoteHostTrusted = false
  10. 2019-11-05 20:57:29,271 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1572958454724_0004
  11. 2019-11-05 20:57:29,271 INFO mapreduce.JobSubmitter: Executing with tokens: []
  12. 2019-11-05 20:57:29,546 INFO conf.Configuration: resource-types.xml not found
  13. 2019-11-05 20:57:29,547 INFO resource.ResourceUtils: Unable to find 'resource-types.xml'.
  14. 2019-11-05 20:57:29,632 INFO impl.YarnClientImpl: Submitted application application_1572958454724_0004
  15. 2019-11-05 20:57:29,671 INFO mapreduce.Job: The url to track the job: http://hadoop02:8088/proxy/application_1572958454724_0004/
  16. # job_时间戳_job的编号(起始 0001)
  17. 2019-11-05 20:57:29,672 INFO mapreduce.Job: Running job: job_1572958454724_0004
  18. 2019-11-05 20:57:39,891 INFO mapreduce.Job: Job job_1572958454724_0004 running in uber mode : false
  19. # 先 map 后 reduce
  20. 2019-11-05 20:57:39,892 INFO mapreduce.Job:  map 0% reduce 0%
  21. 2019-11-05 20:58:08,354 INFO mapreduce.Job:  map 100% reduce 0%
  22. 2019-11-05 20:58:20,518 INFO mapreduce.Job:  map 100% reduce 100%
  23. # 运行成功
  24. 2019-11-05 20:58:21,541 INFO mapreduce.Job: Job job_1572958454724_0004 completed successfully
  25. # 结果计数器：总体统计（读了多少数据，写出多少字节数据）
  26. 2019-11-05 20:58:24,245 INFO mapreduce.Job: Counters: 54
  27.         File System Counters
  28.                 FILE: Number of bytes read=140544
  29.                 FILE: Number of bytes written=1184301
  30.                 FILE: Number of read operations=0
  31.                 FILE: Number of large read operations=0
  32.                 FILE: Number of write operations=0
  33.                 HDFS: Number of bytes read=451992
  34.                 HDFS: Number of bytes written=35843
  35.                 HDFS: Number of read operations=14
  36.                 HDFS: Number of large read operations=0
  37.                 HDFS: Number of write operations=2
  38.                 HDFS: Number of bytes read erasure-coded=0
  39.         Job Counters 
  40.                 Launched map tasks=3
  41.                 Launched reduce tasks=1
  42.                 Data-local map tasks=3
  43.                 Total time spent by all maps in occupied slots (ms)=74817
  44.                 Total time spent by all reduces in occupied slots (ms)=9165
  45.                 Total time spent by all map tasks (ms)=74817
  46.                 Total time spent by all reduce tasks (ms)=9165
  47.                 Total vcore-milliseconds taken by all map tasks=74817
  48.                 Total vcore-milliseconds taken by all reduce tasks=9165
  49.                 Total megabyte-milliseconds taken by all map tasks=76612608
  50.                 Total megabyte-milliseconds taken by all reduce tasks=9384960
  51.         Map-Reduce Framework
  52.                 Map input records=8442
  53.                 Map output records=65712
  54.                 Map output bytes=702105
  55.                 Map output materialized bytes=140556
  56.                 Input split bytes=285
  57.                 Combine input records=65712
  58.                 Combine output records=8943
  59.                 Reduce input groups=2981
  60.                 Reduce shuffle bytes=140556
  61.                 Reduce input records=8943
  62.                 Reduce output records=2981
  63.                 Spilled Records=17886
  64.                 Shuffled Maps =3
  65.                 Failed Shuffles=0
  66.                 Merged Map outputs=3
  67.                 GC time elapsed (ms)=4536
  68.                 CPU time spent (ms)=6180
  69.                 Physical memory (bytes) snapshot=671293440
  70.                 Virtual memory (bytes) snapshot=10928197632
  71.                 Total committed heap usage (bytes)=381550592
  72.                 Peak Map Physical memory (bytes)=189513728
  73.                 Peak Map Virtual memory (bytes)=2733113344
  74.                 Peak Reduce Physical memory (bytes)=106672128
  75.                 Peak Reduce Virtual memory (bytes)=2735411200
  76.         Shuffle Errors
  77.                 BAD_ID=0
  78.                 CONNECTION=0
  79.                 IO_ERROR=0
  80.                 WRONG_LENGTH=0
  81.                 WRONG_MAP=0
  82.                 WRONG_REDUCE=0
  83.         File Input Format Counters 
  84.                 Bytes Read=451707
  85.         File Output Format Counters 
  86.                 Bytes Written=35843

WordCount 程序官方

下载 jar包

  1. <dependency>
  2.   <groupId>org.apache.hadoop</groupId>
  3.   <artifactId>hadoop-mapreduce-examples</artifactId>
  4.   <version>${hadoop.version}</version>
  5. </dependency>

/maven-repo/org/apache/hadoop/hadoop-mapreduce-examples/3.2.1/hadoop-mapreduce-examples-3.2.1.jar!/org/apache/hadoop/examples/WordCount.class

观察依赖

ExampleDriver

这里就是 hadoop jar XXX.jar 主函数

主函数位置写的是 wordcount的原因

  1. package org.apache.hadoop.examples;
  3. import org.apache.hadoop.examples.dancing.DistributedPentomino;
  4. import org.apache.hadoop.examples.dancing.Sudoku;
  5. import org.apache.hadoop.examples.pi.DistBbp;
  6. import org.apache.hadoop.examples.terasort.TeraGen;
  7. import org.apache.hadoop.examples.terasort.TeraSort;
  8. import org.apache.hadoop.examples.terasort.TeraValidate;
  9. import org.apache.hadoop.util.ProgramDriver;
  11. public class ExampleDriver {
  12.   public ExampleDriver() {
  13.   }
  15.   public static void main(String[] argv) 注
  16.     //退出标
  17.     int exitCode = -1;
  18.       //项目驱动对对象，类似一个容器集合
  19.     ProgramDriver pgd = new ProgramDriver();
  21.     try {
  22.       //运行wordcount,实际运行的是 WordCount.class
  23.       pgd.addClass("wordcount", WordCount.class, "A map/reduce program that counts the words in the input files.");
  24.       pgd.addClass("wordmean", WordMean.class, "A map/reduce program that counts the average length of the words in the input files.");
  25.       pgd.addClass("wordmedian", WordMedian.class, "A map/reduce program that counts the median length of the words in the input files.");
  26.       pgd.addClass("wordstandarddeviation", WordStandardDeviation.class, "A map/reduce program that counts the standard deviation of the length of the words in the input files.");
  27.       pgd.addClass("aggregatewordcount", AggregateWordCount.class, "An Aggregate based map/reduce program that counts the words in the input files.");
  28.       pgd.addClass("aggregatewordhist", AggregateWordHistogram.class, "An Aggregate based map/reduce program that computes the histogram of the words in the input files.");
  29.       pgd.addClass("grep", Grep.class, "A map/reduce program that counts the matches of a regex in the input.");
  30.       pgd.addClass("randomwriter", RandomWriter.class, "A map/reduce program that writes 10GB of random data per node.");
  31.       pgd.addClass("randomtextwriter", RandomTextWriter.class, "A map/reduce program that writes 10GB of random textual data per node.");
  32.       pgd.addClass("sort", Sort.class, "A map/reduce program that sorts the data written by the random writer.");
  33.       pgd.addClass("pi", QuasiMonteCarlo.class, "A map/reduce program that estimates Pi using a quasi-Monte Carlo method.");
  34.       pgd.addClass("bbp", BaileyBorweinPlouffe.class, "A map/reduce program that uses Bailey-Borwein-Plouffe to compute exact digits of Pi.");
  35.       pgd.addClass("distbbp", DistBbp.class, "A map/reduce program that uses a BBP-type formula to compute exact bits of Pi.");
  36.       pgd.addClass("pentomino", DistributedPentomino.class, "A map/reduce tile laying program to find solutions to pentomino problems.");
  37.       pgd.addClass("secondarysort", SecondarySort.class, "An example defining a secondary sort to the reduce.");
  38.       pgd.addClass("sudoku", Sudoku.class, "A sudoku solver.");
  39.       pgd.addClass("join", Join.class, "A job that effects a join over sorted, equally partitioned datasets");
  40.       pgd.addClass("multifilewc", MultiFileWordCount.class, "A job that counts words from several files.");
  41.       pgd.addClass("dbcount", DBCountPageView.class, "An example job that count the pageview counts from a database.");
  42.       pgd.addClass("teragen", TeraGen.class, "Generate data for the terasort");
  43.       pgd.addClass("terasort", TeraSort.class, "Run the terasort");
  44.       pgd.addClass("teravalidate", TeraValidate.class, "Checking results of terasort");
  45.       exitCode = pgd.run(argv);
  46.     } catch (Throwable var4) {
  47.       var4.printStackTrace();
  48.     }
  50.     System.exit(exitCode);
  51.   }
  52. }

WordCount

  1. //
  2. // Source code recreated from a .class file by IntelliJ IDEA
  3. // (powered by Fernflower decompiler)
  4. //
  6. package org.apache.hadoop.examples;
  8. import java.io.IOException;
  9. import java.util.Iterator;
  10. import java.util.StringTokenizer;
  11. import org.apache.hadoop.conf.Configuration;
  12. import org.apache.hadoop.fs.Path;
  13. import org.apache.hadoop.io.IntWritable;
  14. import org.apache.hadoop.io.Text;
  15. import org.apache.hadoop.mapreduce.Job;
  16. import org.apache.hadoop.mapreduce.Mapper;
  17. import org.apache.hadoop.mapreduce.Reducer;
  18. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  19. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  20. import org.apache.hadoop.util.GenericOptionsParser;
  22. public class WordCount {
  23.     public WordCount() {
  24.     }
  26.     public static void main(String[] args) throws Exception {
  27.         Configuration conf = new Configuration();
  28.         String[] otherArgs = (new GenericOptionsParser(conf, args)).getRemainingArgs();
  29.         if (otherArgs.length < 2) {
  30.             System.err.println("Usage: wordcount <in> [<in>...] <out>");
  31.             System.exit(2);
  32.         }
  34.         Job job = Job.getInstance(conf, "word count");
  35.         job.setJarByClass(WordCount.class);
  36.         job.setMapperClass(WordCount.TokenizerMapper.class);
  37.         job.setCombinerClass(WordCount.IntSumReducer.class);
  38.         job.setReducerClass(WordCount.IntSumReducer.class);
  39.         job.setOutputKeyClass(Text.class);
  40.         job.setOutputValueClass(IntWritable.class);
  42.         for(int i = 0; i < otherArgs.length - 1; ++i) {
  43.             FileInputFormat.addInputPath(job, new Path(otherArgs[i]));
  44.         }
  46.         FileOutputFormat.setOutputPath(job, new Path(otherArgs[otherArgs.length - 1]));
  47.         System.exit(job.waitForCompletion(true) ? 0 : 1);
  48.     }
  50.     public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
  51.         private IntWritable result = new IntWritable();
  53.         public IntSumReducer() {
  54.         }
  56.         public void reduce(Text key, Iterable<IntWritable> values, Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {
  57.             int sum = 0;
  59.             IntWritable val;
  60.             for(Iterator var5 = values.iterator(); var5.hasNext(); sum += val.get()) {
  61.                 val = (IntWritable)var5.next();
  62.             }
  64.             this.result.set(sum);
  65.             context.write(key, this.result);
  66.         }
  67.     }
  69.     public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> {
  70.         private static final IntWritable one = new IntWritable(1);
  71.         private Text word = new Text();
  73.         public TokenizerMapper() {
  74.         }
  76.         public void map(Object key, Text value, Mapper<Object, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException {
  77.             StringTokenizer itr = new StringTokenizer(value.toString());
  79.             while(itr.hasMoreTokens()) {
  80.                 this.word.set(itr.nextToken());
  81.                 context.write(this.word, one);
  82.             }
  84.         }
  85.     }
  86. }

编程结构

  2. public class WordCount {
  3.   public WordCount() {
  4.   }
  5.     //主方法入口
  6.   public static void main(String[] args) throws Exception {
  7.     //组装map和reduce   并进行运行提交...
  8.   }
  10.   public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
  11.         //map方法....
  12.   }
  14.   public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> {
  15.     //reduce的方法...
  16.   }
  17. }

自己写 MapReduce

Driver.java :组装 map 和 reduce

  1. package pers.sfl.mapreduce;
  3. import java.io.IOException;
  5. import org.apache.hadoop.conf.Configuration;
  6. import org.apache.hadoop.fs.Path;
  7. import org.apache.hadoop.io.IntWritable;
  8. import org.apache.hadoop.io.Text;
  9. import org.apache.hadoop.mapreduce.Job;
  10. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  11. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  14. /**
  15.  * 驱动类
  16.  * @author Administrator
  17.  *
  18.  */
  19. public class Driver {
  20.     public static void main(String[] args) {
  21.         //将mapper  reducer类进行一个封装  封装为一个任务----job（作业）
  22.         //加载配置文件
  23.         Configuration conf=new Configuration();
  24.         //启动一个Job  创建一个job对象
  25.         try {
  26.             Job job=Job.getInstance(conf);
  27.             //设置这个job
  28.             //设置整个job的主函数入口
  29.             job.setJarByClass(Driver.class);
  31.             //设置job的mappper的类
  32.             job.setMapperClass(WordCountMapper.class);
  34.             //设置job的reducer的类
  35.             job.setReducerClass(WordCountReducer.class);
  38.             //设置map输出key   value的类型
  39.             //指定了泛型  这里为什么还要设置一次   泛型的作用范围  编译的时候生效   运行的时候泛型会自动擦除
  40.             job.setMapOutputKeyClass(Text.class);
  41.             job.setMapOutputValueClass(IntWritable.class);
  43.             //设置reduce的输出的k   v类型  以下方法设置的是mr的最终输出
  44.             job.setOutputKeyClass(Text.class);
  45.             job.setOutputValueClass(IntWritable.class);
  48.             //指定需要统计的文件的输入路径  FileInputFormat  文件输入类
  49.             Path inpath=new Path(args[0]);
  50.             FileInputFormat.addInputPath(job, inpath);
  52.             //指定输出目录  输出路径不能存在的  否则会报错  默认输出是覆盖式的输出  如果输出目录存在  有可能造成原始数据的丢失
  53.             Path outpath=new Path(args[1]);
  54.             FileOutputFormat.setOutputPath(job, outpath);
  56.             //提交job  执行这一句的时候 job才会提交  上面做的一系列的工作  都是设置job
  57.             //job.submit();
  58.             job.waitForCompletion(true);
  60.         } catch (Exception e) {
  62.             e.printStackTrace();
  63.         }
  65.     }
  67. }

WordCountMapper.java

  1. package pers.sfl.mapreduce;
  2. import java.io.IOException;
  3. import java.io.Serializable;
  5. import org.apache.hadoop.io.IntWritable;
  6. import org.apache.hadoop.io.LongWritable;
  7. import org.apache.hadoop.io.Text;
  8. /**
  9.  * 单词统计
  10.  */
  11. import org.apache.hadoop.mapreduce.Mapper;
  12. /**
  13.  * 输入和输出   map类型
  14.  * KEYIN, 输入的键的类型   这里指的是每一行的起始偏移量  long  0   12
  15.  * VALUEIN,输入的value的类型  这里指的是每一行的内容   和偏移量一一对应的   String
  16.  * 输出的类型取决于  业务
  17.  *  KEYOUT,  输出的键的类型  这里指的每一个单词   ---  string
  18.  *  VALUEOUT，输出的值的类型  这里指的单词的次数  ----  int
  19.  * @author Administrator
  20.  * 
  21.  * 
  22.  * 这里的数据类型  不能使用java的原生类型
  23.  *     序列化：数据持久化存储  或  网络传输的时候  数据需要序列化和反序列化的
  24.  *     张三---序列化------010101110-----反序列化-----张三
  25.  * java-----Serializable
  26.  * mapreduce编程中的用于传输的数据类型必须是序列化和反序列化能力的
  27.  *   hadoop弃用了java中原生的Serializable   实现的自己的一套序列化和反序列化的 接口Writable  只会对数据的值进行序列化和反序列化
  28.  *   原因：java中的序列化和反序列化太重    繁琐
  29.  *   Long   1   
  30.  *   对于一些常用的数据类型  hadoop帮我们实现好了：
  31.  *   int------intWritable
  32.  *   long----LongWritable
  33.  *   string-----Text
  34.  *   byte------ByteWritable
  35.  *   double----DoubleWritable
  36.  *   float-----FloatWritable
  37.  *   boolean-----BooleanWritable
  38.  *   null-----NullWritable
  39.  *自己定义的需要序列化和反序列化  实现  Writable接口
  40.  */
  41. public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{
  42.     //重写map函数
  43.     /**
  44.      * 参数：
  45.      * hadoop的底层数据读取的时候   字节读取的
  46.      * LongWritable key：输入的key   这里指的是每一行的偏移量   没有实际作用  一行的标识而已
  47.      * Text value：输入的value  这里指的是一行的内容
  48.      * Context context：上下文对象   用于传输     传输reduce中
  49.      * 函数的调用频率：
  50.      *     一行调用一次
  51.      * 如果一个文件  10行-----map函数会被调用10次
  52.      */
  53.     @Override
  54.     protected void map(LongWritable key, 
  55.             Text value, 
  56.             Context context)
  57.             throws IOException, InterruptedException {
  61.         //创建一个流   进行读取（mapreduce框架帮你做了）    每一行内容进行切分
  62.         //获取每一行内容  进行切分
  63.         //text--toString()--String
  64.         String line = value.toString();
  65.         //进行切分   hello    word    hello    ww
  66.         String[] words = line.split("\t");
  67.         //循环遍历每一个单词  进行统计    直接发送到reduce端   发送的时候  k-v
  68.         for(String w:words){
  69.             //将String---Text
  70.             Text mk=new Text(w);
  71.             //int----IntWritable
  72.             IntWritable mv=new IntWritable(1);
  73.             //hello,1    world,1     hello,1    ww,1
  74.             //这里write是直接写出  调用一次  就会写出一个   k---v写出reduce端
  75.             context.write(mk, mv);
  76.         }
  78.     }
  80. }

WordCountReduce.java

  1. package pers.sfl.mapreduce;
  2. import java.io.IOException;
  4. import org.apache.hadoop.io.IntWritable;
  5. import org.apache.hadoop.io.Text;
  6. /**
  7.  * 单词统计
  8.  */
  9. import org.apache.hadoop.mapreduce.Reducer;
  10. /**
  11.  * reduce的数据来源于map
  12.  * @author Administrator
  13.  *KEYIN,  输入的key的类型    这里指的是map输出key类型   Text
  14.  * VALUEIN, 输入的value的类型  这里指的是map输出的value的类型    IntWritable
  15.  * 
  16.  * 输出
  17.  *KEYOUT, 输出的key    这里指的单词的类型  Text
  18.  *VALUEOUT，输出的value的类型   这里指的是单词的总次数    IntWritable
  19.  */
  20. public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable>{
  21.     //重写  reduce方法
  22.     /**
  23.      * 到reduce端的数据  是已经分好组的数据
  24.      * 默认情况下   按照map输出的key进行分组  将map输出的key相同的分为一组
  25.      * Text key,    每一组中的相同的key
  26.      * Iterable<IntWritable> values,  每一组中的所有的value值  封装到一个迭代器中了
  27.         Context context：上下文对象   用于传输的  写出到hdfs中
  29.         reduce调用频率：
  30.             一组调用一次  每次只统计一个单词最终结果
  31.         每一组只能访问本组的数据  没有办法和上一组的额数据  下一组的数据共享的
  32.      */
  33.     @Override
  34.     protected void reduce(Text key, Iterable<IntWritable> values,
  35.             Context context) throws IOException, InterruptedException {
  36.         //循环遍历values  求和
  37.         int sum=0;
  38.         for(IntWritable v:values){
  39.             //intwritable---int  数值类型   get   将hadoop中的类型转换为java中的类型
  40.             sum+=v.get();
  41.         }
  42.         //写出结果文件
  43.         IntWritable rv=new IntWritable(sum);
  44.         context.write(key, rv);
  45.     }
  47. }

jar上传运行

将编写好的程序打成jar包,上传到服务器任意一个节点

运行

运行 指定main函数，程序中指定路径， 没有运行参数的:hadoop jar wc01.jar

  1. 报错：Error: java.io.IOException: Initialization of all the collectors failed. Error in last collector was :class com.sun.jersey.core.impl.provider.entity.XMLJAXBElementProvider$Text
  3. Text导包问题
  5. 运行结果：2个文件
  6. _SUCCESS:大小为0   运行标志文件   证明job运行成功
  7. part-r-00000：结果文件
  8.   r:最终结果reduce输出的
  9.   m:最终结果map输出的结果
  10.   00000：文件编号   从0开始
  11.     输出多个文件则顺序递增

没有指定main函数：

  1. 主函数的类一定是 全限定名
  2.   hadoop jar wc02_nomain.jar pers.sfl.mapreduce.Driver

没有主函数 又需要参数的

  1. hadoop jar wc03_nomain_pa.jar  pers.sfl.mapreduce.Driver /wcin /wc_out02

说明

文件输入(加载的类)：FileInputFormat,默认的实现类是：TextInputFormat

文件读取： 默认的文件读取器:RecordReader，默认实现类：LineRecordReader

LineRecordReader的核心方法

  1. //判断当前的文件是否还有下一个 k-v 要读取
  2. public boolean nextKeyValue() throws IOException {
  3. }
  5. @Override
  6. //获取当前行返回的偏移量 -- LongWritable
  7. public LongWritable getCurrentKey() {
  8.   return key;
  9. }
  11. @Override
  12. //获取当前行的内容 -- Text
  13. public Text getCurrentValue() {
  14.   return value;
  15. }

过程

  1. 通过文件加载器和读取器，将文件内容发送到Map 端
  2. Map 端实现map方法(LongWritable key, Text value, Context context)
  3. 将内容进行切割得到(helllo,1),(helllo,1),(helllo,1),(world,1)

  4. shuffle 过程

    • 分区：partitioner
    • 排序： sort按照 key进行排序(字典顺序) writablecomparable
    • 分组：grop相同的 key 分为一组 writablecomparator
    • combiner 组件：优化
  5. reduce接收到的是key相同的一组数据
  6. FileInputFormat -> TextOutputFormat 文件输出类
  7. 使用文件写出器RecordWriter(输出流) -> LineRecordWriter 写到 hdfs

mapreduce的编程套路

  • reduce: 计算统计
  • 一般map端:只做数据的字段的摘取

3个文件中最大值 最小值 平均值

  1. maptask：
  2.     每次只能取到一行的内容
  3.     拿到每一个数值  ->  发送
  4.     map：输出   
  5.         如何判断key：根据reduce端如何接受
  6.         map输出的key：分组
  7.         想按找哪一个字段进行分组  map的key就是那一个字段
  8.         reduce一组会调用一次
  10.         key:“数据”  Text
  11.         value:数据的值
  13. reducetask:一组
  14.     接受的数据  所有的数据分到一组
  15.     一次性访问所有的数据'
  17.     将所有的数据取出--values  求最大值  最小值  平均值