第一章.shuffle机制

1.WritableComparable排序

  1. # 排序概述
  2. - 排序是MapReduce框架中最重要的操作之一
  3. - MapTask和ReduceTask均会对数据按照key进行排序该操作属于hadoop的默认行为,任何应用程序中的数据均会被排序
  4.   而不管逻辑上是否需要
  5. - 默认排序是按照字典顺序排序,且实现该排序的方法是快速排序
  6. - 对于MapTask,它会将处理的结果暂时放在环形缓冲区内,当环形缓冲区使用率达到一定阈值后,再对缓冲区中的数据进行一次快速排序,并将这些数据溢写到磁盘上,而当数据处理完毕后,它会对磁盘上的所有文件进行归并排序
  7. - 对于ReduceTask,它从每个MapTask上远程拷贝相应的数据文件,如果文件大小超过一定阈值,则溢写磁盘上,否则储存在内存中,如果磁盘上文件数目达到一定阈值,则进行一次归并排序以生成一个更大文件如果内存中文件大小或者数目超过一定阈值,则进行一次合并后将数据溢写到磁盘上,当所有的数据拷贝完毕后,ReduceTask统一对内存和磁盘上的所有数据进行一次归并排序

案例实操: 统计文件中手机号,上行流量,下行流量,总流量,并再对总流量进行排序

$07[Hadoop-MapReduce框架原理(下)] - 图1

代码实现

  1. FlowBean对象在之前的基础上增加了比较功能
  1. package com.atguigu.mr.comparable;
  4. import org.apache.hadoop.io.WritableComparable;
  6. import java.io.DataInput;
  7. import java.io.DataOutput;
  8. import java.io.IOException;
  10. /*
  11. 自定义的类的对象作为key
  12. 1.自定义的类 需要实现WritableComparable接口
  13. 2. WritableComparable继承了Writable和Comparable
  14. 3.重写Writable和Comparable接口中的抽象方法
  15.  */
  16. public class FlowBean implements WritableComparable<FlowBean> {
  17.     private long upFlow;
  18.     private long downFlow;
  19.     private long sumFlow;
  21.     public FlowBean(){
  23.     }
  25.     public FlowBean(long upFlow, long downFlow,long sumFlow) {
  26.         this.upFlow=upFlow;
  27.         this.downFlow=downFlow;
  28.         this.sumFlow =sumFlow;
  29.     }
  30.     /**
  31.     * @Description: 当序列化时调用此方法
  32.     * @Param: [dataOutput]
  33.     * @return: void
  34.     * @Author: jcsune
  35.     * @Date: 2021/7/12
  36.     */
  37.     @Override
  38.     public void write(DataOutput out) throws IOException {
  39.         out.writeLong(upFlow);
  40.         out.writeLong(downFlow);
  41.         out.writeLong(sumFlow);
  42.     }
  43.     /**
  44.     * @Description: 当反序列化时调用此方法,注意反序列化时读取数据的顺序要和序列化时写数据的顺序保持一致
  45.     * @Param: [in]
  46.     * @return: void
  47.     * @Author: jcsune
  48.     * @Date: 2021/7/12
  49.     */
  50.     @Override
  51.     public void readFields(DataInput in) throws IOException {
  52.        upFlow = in.readLong();
  53.        downFlow = in.readLong();
  54.        sumFlow = in.readLong();
  56.     }
  57.     /*
  58.     在该方法中指定按照哪个属性进行排序
  59.      */
  60.     @Override
  61.     public int compareTo(FlowBean o){
  62.         //按照总流量进行排序
  63.         return Long.compare(this.sumFlow,o.sumFlow);
  64.     }
  66.     public long getUpFlow() {
  67.         return upFlow;
  68.     }
  70.     public void setUpFlow(long upFlow) {
  71.         this.upFlow = upFlow;
  72.     }
  74.     public long getDownFlow() {
  75.         return downFlow;
  76.     }
  78.     public void setDownFlow(long downFlow) {
  79.         this.downFlow = downFlow;
  80.     }
  82.     public long getSumFlow() {
  83.         return sumFlow;
  84.     }
  86.     public void setSumFlow(long sumFlow) {
  87.         this.sumFlow = sumFlow;
  88.     }
  90.     @Override
  91.     public String toString() {
  92.         return upFlow + " " + downFlow + " " +sumFlow;
  93.     }
  94. }
  1. Mapper类
  1. package com.atguigu.mr.comparable;
  3. import org.apache.hadoop.io.LongWritable;
  4. import org.apache.hadoop.io.Text;
  5. import org.apache.hadoop.mapreduce.Mapper;
  7. import java.io.IOException;
  8. /*
  9.     注意:一定要让FlowBean作为key,因为只能对key进行排序
  10.  */
  12. public class FlowMapper extends Mapper<LongWritable,Text,FlowBean,Text> {
  14.     @Override
  15.     protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
  16.         //将value转成字符串
  17.         String line = value.toString();
  18.         //将数据切割
  19.         String[] flowInfo = line.split("\t");
  20.         //封装k,v
  21.         FlowBean outkey =new FlowBean(Long.parseLong(flowInfo[1]),Long.parseLong(flowInfo[2]),Long.parseLong(flowInfo[3]));
  22.         Text outValue = new Text(flowInfo[0]);
  23.         //写出k,v
  24.         context.write(outkey,outValue);
  26.     }
  27. }
  1. Reducer类
  1. package com.atguigu.mr.comparable;
  3. import org.apache.hadoop.io.Text;
  4. import org.apache.hadoop.mapreduce.Reducer;
  6. import java.io.IOException;
  9. /*
  10. 只是将k,v进行调换
  11. 注意:一定要将泛型调换顺序
  13.  */
  14. public class FlowReducer extends Reducer< FlowBean,Text,Text, FlowBean> {
  17.     @Override
  18.     protected void reduce(FlowBean key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
  19.        //遍历所有的value
  20.         for(Text outKey: values){
  21.             //将k,v写出去(原来的k变成v,原来的v变成k)
  22.             context.write(outKey,key);
  23.         }
  24.     }
  25. }
  1. Driver类
  1. package com.atguigu.mr.comparable;
  3. import org.apache.hadoop.conf.Configuration;
  4. import org.apache.hadoop.fs.Path;
  5. import org.apache.hadoop.io.Text;
  6. import org.apache.hadoop.mapreduce.Job;
  7. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  8. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  10. public class FlowDriver {
  11.     public static void main(String[] args) throws Exception{
  12.         //创建Job对象
  13.         Job job = Job.getInstance(new Configuration());
  14.         //给Job对象的属性赋值
  16.         job.setMapperClass(FlowMapper.class);
  17.         job.setReducerClass(FlowReducer.class);
  18.         //2.3设置Mapper输出的k,v的类型
  19.         job.setMapOutputKeyClass(FlowBean.class);
  20.         job.setMapOutputValueClass(Text.class);
  21.         //2.4设置最终输出的K,V的类型（在这是Reducer输出的K,V类型）
  22.         job.setOutputKeyClass(Text.class);
  23.         job.setOutputValueClass(FlowBean.class);
  24.         //2.5设置数据的输入和输出路径
  25.         FileInputFormat.setInputPaths(job,new Path("C:\\io\\input5"));
  26.         //注意：输出路径不能存在！！！！！！！！！！！！
  27.         FileOutputFormat.setOutputPath(job,new Path("C:\\io\\output5"));
  29.         //3.提交Job对象（执行Job）
  30.         /*
  31.             参数 ：是否打印进度
  32.             返回值 ：如果是true则表示job执行成功
  33.          */
  34.         job.waitForCompletion(true);
  35.     }
  36. }
  1. 执行程序,查看结果

$07[Hadoop-MapReduce框架原理(下)] - 图2

2.WritableComparable扩展(区内排序)

需求:要求每个省份手机号输出的文件中按照总流量内部排序

  1. 增加自定义分区类
  1. package com.atguigu.mr.comparable2;
  3. import org.apache.hadoop.io.Text;
  4. import org.apache.hadoop.mapreduce.Partitioner;
  6. /*
  7. 自定义分区类
  8. key:mapper输出的key
  9. value:mapper输出的value
  10.  */
  11. public class MyPartitioner extends Partitioner<FlowBean, Text> {
  12.     @Override
  13.     public int getPartition(FlowBean flowBean, Text text, int numPartitions) {
  14.         //将Text转成String
  15.         String phoneNumber = text.toString();
  16.         //匹配手机号
  17.         if(phoneNumber.startsWith("136")){
  18.             return 0;
  19.         }else if(phoneNumber.startsWith("137")){
  20.             return 1;
  21.         }else if(phoneNumber.startsWith("137")){
  22.             return 2;
  23.         }else if(phoneNumber.startsWith("138")){
  24.             return 3;
  25.         }else {
  26.             return 4;
  27.         }
  28.     }
  29. }
  1. 在驱动类中添加分区类
  1. package com.atguigu.mr.comparable2;
  3. import org.apache.hadoop.conf.Configuration;
  4. import org.apache.hadoop.fs.Path;
  5. import org.apache.hadoop.io.Text;
  6. import org.apache.hadoop.mapreduce.Job;
  7. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  8. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  10. public class FlowDriver {
  11.     public static void main(String[] args) throws Exception{
  12.         //创建Job对象
  13.         Job job = Job.getInstance(new Configuration());
  14.         //设置使用自定义分区类
  15.         job.setPartitionerClass(MyPartitioner.class);
  16.         //设置ReduceTask的数量
  17.         job.setNumReduceTasks(5);
  18.         //给Job对象的属性赋值
  20.         job.setMapperClass(FlowMapper.class);
  21.         job.setReducerClass(FlowReducer.class);
  22.         //2.3设置Mapper输出的k,v的类型
  23.         job.setMapOutputKeyClass(FlowBean.class);
  24.         job.setMapOutputValueClass(Text.class);
  25.         //2.4设置最终输出的K,V的类型（在这是Reducer输出的K,V类型）
  26.         job.setOutputKeyClass(Text.class);
  27.         job.setOutputValueClass(FlowBean.class);
  28.         //2.5设置数据的输入和输出路径
  29.         FileInputFormat.setInputPaths(job,new Path("C:\\io\\input6"));
  30.         //注意：输出路径不能存在！！！！！！！！！！！！
  31.         FileOutputFormat.setOutputPath(job,new Path("C:\\io\\output6"));
  33.         //3.提交Job对象（执行Job）
  34.         /*
  35.             参数 ：是否打印进度
  36.             返回值 ：如果是true则表示job执行成功
  37.          */
  38.         job.waitForCompletion(true);
  39.     }
  40. }

$07[Hadoop-MapReduce框架原理(下)] - 图3

3.Combiner合并及案例实现

  1. - Combiner是MR程序中MR程序中Mapper和Reducer之外的一种组件
  2. - Combiner组件的父类就是Reducer
  3. - Combiner与Reducer的区别在于运行的位置
  4.   - Combiner是在每一个MapTask所在的节点运行
  5.   - Reducer是接收全局所有Mapper的输出结果
  6. - Combiner的意义就是对每一个MapTask的输出进行局部汇总,以减少网络传输量
  7. - Combiner能够应用的前提下是不能影响最终的业务逻辑

案例实操: 统计过程中对每一个MapTask的输出进行局部汇总,以减少网络传输量即采用Combiner功能

未使用Combiner

$07[Hadoop-MapReduce框架原理(下)] - 图4

使用Combiner

  1. package com.atguigu.mr.combiner;
  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.output.FileOutputFormat;
  11. public class WCDriver {
  12.     public static void main(String[] args) throws Exception{
  13.         //1.创建job对象
  14.         Configuration conf = new Configuration();//可以在该对象中做一些设置
  15.         Job job = Job.getInstance();
  16.         //设置Combiner
  17.         job.setCombinerClass(WCCombiner.class);
  18.         //2.给job对象的属性赋值
  19.         //设置jar加载路径(如果是本地模式,可以不设置)
  20.         job.setJarByClass(WCDriver.class);
  22.         //设置Mapper和Reduce类
  23.         job.setMapperClass(WCMapper.class);
  24.         job.setReducerClass(WCReducer.class);
  25.         //设置Mapper输出的k,v类型
  26.         job.setMapOutputKeyClass(Text.class);
  27.         job.setMapOutputValueClass(IntWritable.class);
  28.         //设置最终输出的K,V类型
  29.         job.setOutputKeyClass(Text.class);
  30.         job.setOutputValueClass(IntWritable.class);
  31.         //设置数据的输入和输出路径
  32.         FileInputFormat.setInputPaths(job,new Path("C:\\io\\input7"));
  33.         //注意输出路径不能存在
  34.         FileOutputFormat.setOutputPath(job,new Path("C:\\io\\output7"));
  35.         //3.提交job对象
  36.         /*
  37.         参数:是否打印进度
  38.         返回值:如果是true则表示job执行成功
  39.          */
  40.         boolean b = job.waitForCompletion(true);
  41.         System.exit(b ? 0 : 1);
  42.     }
  43. }

$07[Hadoop-MapReduce框架原理(下)] - 图5

4.outputFormat数据输出

  1. # outputFormat接口实现类
  2. - outputFormat是MapReduce输出的基类,所有实现MapReduce输出都实现了OutputFormat接口
  3. - 文本输出TextOutputFormat:默认的输出格式是TextOutputFormat,它把每条记录写为文本行,它的键和值可以是任意类型,因为TextOutputFormat调用toString()方法把他们转换为字符串
  4. - SequenceFileOutputFormat:将SequerceFileOutputFormat输出作为后续MapReduce任务的输入,这便是好的一种输出格式,因为它的格式紧凑,很容易被压缩
  5. - 自定义OutputFormat
  6. 1. 自定义一个类继承FileOutputFormat
  7. 2. 改写RecordWriter,具体改写输出数据的方法write()

案例实操

  1. 需求:过滤输入的log日志,包含atguigu的网站输出到atguigu.txt,不包含atguigu的网站输出到other.txt
  2. 需求分析

$07[Hadoop-MapReduce框架原理(下)] - 图6

MyRecordWriter.java

  1. package com.atguigu.mr.output;
  3. import org.apache.hadoop.fs.FSDataOutputStream;
  4. import org.apache.hadoop.fs.FileSystem;
  5. import org.apache.hadoop.fs.Path;
  6. import org.apache.hadoop.io.IOUtils;
  7. import org.apache.hadoop.io.LongWritable;
  8. import org.apache.hadoop.io.Text;
  9. import org.apache.hadoop.mapreduce.RecordWriter;
  10. import org.apache.hadoop.mapreduce.TaskAttemptContext;
  11. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  13. import java.io.IOException;
  16. /*
  17. 自定义RecordWriter类
  18.  */
  19. public class MyRecordWriter extends RecordWriter<LongWritable, Text> {
  20.     private FSDataOutputStream atguigu;
  21.     private FSDataOutputStream other;
  23. /*
  24. 创建流
  25.  */
  26.     public MyRecordWriter(TaskAttemptContext job) {
  27.        try{
  28.            //1.创建流
  29.            FileSystem fs = FileSystem.get(job.getConfiguration());
  30.            //2.创建流--输出流
  31.            //获取输出路径
  32.            Path outputPath = FileOutputFormat.getOutputPath(job);
  33.            atguigu =fs.create(new Path(outputPath,"atguigu.txt"));
  34.            other = fs.create(new Path(outputPath,"other.txt"));
  36.        }catch (Exception e){
  37.            //终止程序运行
  38.            throw new RuntimeException(e.getMessage());
  39.        }
  40.     }
  41.     /*
  42.     写数据
  43.      */
  44.     @Override
  45.     public void write(LongWritable key, Text value) throws IOException, InterruptedException {
  46.         //先判断网址有没有包含atguigu
  47.         String address = value.toString() + "\n";
  48.         if(address.contains("atguigu")){
  49.             //写到atguigu.txt
  50.             atguigu.write(address.getBytes());
  51.         }else{
  52.             //写到other.txt
  53.             other.write(address.getBytes());
  54.         }
  56.     }
  57.     /*
  58.     在该方法中用来关闭资源---当写操作结束以后才会调用此方法(框架调用的)
  59.      */
  60.     @Override
  61.     public void close(TaskAttemptContext context) throws IOException, InterruptedException {
  62.         IOUtils.closeStream(atguigu);
  63.         IOUtils.closeStream(other);
  64.     }
  65. }

MyOutputFormat.java

  1. package com.atguigu.mr.output;
  3. import org.apache.hadoop.io.LongWritable;
  4. import org.apache.hadoop.io.Text;
  5. import org.apache.hadoop.mapreduce.RecordWriter;
  6. import org.apache.hadoop.mapreduce.TaskAttemptContext;
  7. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  9. import java.io.IOException;
  11. public class MyOutputFormat extends FileOutputFormat<LongWritable,Text> {
  13.     /*
  14.     自定义OutputFormat类
  15.     1. 自定义一个类并继承FileOutputFormat,因为继承此类只需要重写getRecordWriter方法而此方法正是我们所需要的
  16.     2. FileOutputFormat泛型的类型Reducer输出的K,V类型
  17.     没有Reducer是Mapper输出的K,V
  18.     没有Mapper也没有那么是Inputformat输出的K,V类型
  19.      */
  21.     @Override
  22.     public RecordWriter<LongWritable, Text> getRecordWriter(TaskAttemptContext job) throws IOException, InterruptedException {
  23.         return new MyRecordWriter(job);
  24.     }
  25. }

OutputDriver.java

  1. package com.atguigu.mr.output;
  3. import org.apache.hadoop.conf.Configuration;
  4. import org.apache.hadoop.fs.Path;
  5. import org.apache.hadoop.io.LongWritable;
  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.output.FileOutputFormat;
  11. public class OutputDriver {
  12.     public static void main(String[] args) throws Exception {
  13.         Job job = Job.getInstance(new Configuration());
  14.         job.setOutputKeyClass(LongWritable.class);
  15.         job.setOutputValueClass(Text.class);
  16.         //设置自定义的outputFormat类--如果不指定使用TextOutputFormat
  17.         job.setOutputFormatClass(MyOutputFormat.class);
  18.         FileInputFormat.setInputPaths(job,new Path("C:\\io\\input8"));
  19.         FileOutputFormat.setOutputPath(job,new Path("C:\\io\\output8"));
  20.         job.waitForCompletion(true);
  21.     }
  22. }

输出结果

$07[Hadoop-MapReduce框架原理(下)] - 图7

5.Join多种应用

1. ReduceJoin工作原理

  1. - Map端的主要工作:为来自不同表或文件的key/value对,打标签以区别不同的来源记录,然后用连接字段作为key,其余部分和新加的标志作为value,最后进行输出
  2. - Reduce端的主要工作:在Reduce端以连接字段作为key的分组已经完成.我们只需要在每一个分组当中将那些来源于不同文件的记录(在Map阶段已经打标志分开,)最后进行合并就OK了

2. 案例实操

$07[Hadoop-MapReduce框架原理(下)] - 图8

$07[Hadoop-MapReduce框架原理(下)] - 图9

MyCompartor.java

  1. package com.atguigu.mr.reducejoin;
  3. import org.apache.hadoop.io.WritableComparable;
  4. import org.apache.hadoop.io.WritableComparator;
  6. /*
  7.     自定义分组
  8.     1.自定义一个类并继承WritableComparator
  9.     2.调用父类的有参构造器
  10.     3.重写compare方法,并在该方法中指定分组的方式
  11.  */
  12. public class MyCompartor extends WritableComparator {
  13.     /*
  14.     WritableComparator(Class<? extends WritableComparable> keyClass,
  15.       boolean createInstances)
  16.       keyClass : Key对象所在的类的运行时类的对象
  17.       createInstances : 是否实例化
  18.      */
  19.     public MyCompartor(){
  20.         super(OrderBean.class,true);
  21.     }
  23.     /*
  24.     重写compare方法,并在该方法中指定分组的方式
  25.     作用:在这里按照pid 进行分组
  26.      */
  27.     @Override
  28.     public int compare(WritableComparable a, WritableComparable b) {
  29.         //向下转型
  30.         OrderBean o1 = (OrderBean) a;
  31.         OrderBean o2 = (OrderBean) b;
  32.         return o1.getPid().compareTo(o2.getPid());
  33.     }
  34. }

OrderBean.java

  1. package com.atguigu.mr.reducejoin;
  3. import org.apache.hadoop.io.WritableComparable;
  5. import java.io.DataInput;
  6. import java.io.DataOutput;
  7. import java.io.IOException;
  9. public class OrderBean implements WritableComparable<OrderBean> {
  10.     private String id;
  11.     private String pid;
  12.     private String amount;
  13.     private String pname;
  15.     public OrderBean(){
  17.     }
  18.     public OrderBean(String id,String pid,String amount,String pname){
  19.         this.id = id;
  20.         this.pid = pid;
  21.         this.amount = amount;
  22.         this.pname = pname;
  24.     }
  26.     public String getId() {
  27.         return id;
  28.     }
  30.     public void setId(String id) {
  31.         this.id = id;
  32.     }
  34.     public String getPid() {
  35.         return pid;
  36.     }
  38.     public void setPid(String pid) {
  39.         this.pid = pid;
  40.     }
  42.     public String getAmount() {
  43.         return amount;
  44.     }
  46.     public void setAmount(String amount) {
  47.         this.amount = amount;
  48.     }
  50.     public String getPname() {
  51.         return pname;
  52.     }
  54.     public void setPname(String pname) {
  55.         this.pname = pname;
  56.     }
  57.     /** 
  58.     * @Description: 排序,先按照pid进行排序,pid相同再按照pname记性排序
  59.     * @Param: [o] 
  60.     * @return: int 
  61.     * @Author: jcsune
  62.     * @Date: 2021/7/15 
  63.     */ 
  64.     @Override
  65.     public int compareTo(OrderBean o) {
  66.         int pidValue = this.pid.compareTo(o.pid);
  67.         if(pidValue==0){//说明pid相同,在按照pname排序
  68.             return -this.pname.compareTo(o.pname);
  69.         }
  70.         return pidValue;
  71.     }
  72.     /**
  73.     * @Description: 序列化
  74.     * @Param: [dataOutput]
  75.     * @return: void
  76.     * @Author: jcsune
  77.     * @Date: 2021/7/15
  78.     */
  79.     @Override
  80.     public void write(DataOutput dataOutput) throws IOException {
  81.               dataOutput.writeUTF(id);
  82.               dataOutput.writeUTF(pid);
  83.               dataOutput.writeUTF(pname);
  84.               dataOutput.writeUTF(amount);
  85.     }
  86.     /**
  87.     * @Description: 反序列化
  88.     * @Param: [dataInput]
  89.     * @return: void
  90.     * @Author: jcsune
  91.     * @Date: 2021/7/15
  92.     */
  93.     @Override
  94.     public void readFields(DataInput dataInput) throws IOException {
  95.             id = dataInput.readUTF();
  96.             pid = dataInput.readUTF();
  97.             pname = dataInput.readUTF();
  98.             amount = dataInput.readUTF();
  99.     }
  101.     @Override
  102.     public String toString() {
  103.         return id + " " + pid + " " + amount + " " + pname;
  104.     }
  105. }

RJmapper.java

  1. package com.atguigu.mr.reducejoin;
  3. import org.apache.hadoop.io.LongWritable;
  4. import org.apache.hadoop.io.NullWritable;
  5. import org.apache.hadoop.io.Text;
  6. import org.apache.hadoop.mapreduce.Mapper;
  7. import org.apache.hadoop.mapreduce.lib.input.FileSplit;
  9. import java.io.IOException;
  11. public class RJMapper extends Mapper<LongWritable, Text,OrderBean, NullWritable> {
  12.     private String fileName;
  13.     /*
  14.     该方法在任务开始的时候只执行一次,在map方法调用前执行
  15.     初始化
  16.      */
  18.     @Override
  19.     protected void setup(Context context) throws IOException, InterruptedException {
  20.        //获取切片信息
  21.         FileSplit fs = (FileSplit) context.getInputSplit();
  22.         //通过切片获取文件名
  23.         fileName = fs.getPath().getName();
  24.     }
  26.     @Override
  27.     protected void cleanup(Context context) throws IOException, InterruptedException {
  28.         super.cleanup(context);
  29.     }
  31.     @Override
  32.     protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
  33.         String line = value.toString();
  34.         //切割数据
  35.         String[] info = line.split("\t");
  36.         //创建对象
  37.         OrderBean bean = new OrderBean();
  38.         //封装
  39.         if("order.txt".equals(fileName)){
  40.             bean.setId(info[0]);
  41.             bean.setPid(info[1]);
  42.             bean.setAmount(info[2]);
  43.             //注意:一定要给Pname一个空串,因为排序的时候先按照pid排序再按照pname排序
  44.             bean.setPname("");
  46.         }
  47.         if("pd.txt".equals(fileName)){
  48.             bean.setPid(info[0]);
  49.             bean.setPname(info[1]);
  50.             bean.setId("");
  51.             bean.setAmount("");
  52.         }
  53.         //写出K,V
  54.         context.write(bean,NullWritable.get());
  55.     }
  56. }

RJReducer.java

  1. package com.atguigu.mr.reducejoin;
  3. import org.apache.hadoop.io.NullWritable;
  4. import org.apache.hadoop.mapreduce.Reducer;
  6. import java.io.IOException;
  7. import java.util.Iterator;
  9. public class RJReducer extends Reducer<OrderBean, NullWritable,OrderBean,NullWritable> {
  10.     @Override
  11.     protected void reduce(OrderBean key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
  12.         //获取迭代器对象
  13.         Iterator<NullWritable> iterator = values.iterator();
  14.         //获取该数据中的第一行
  15.         iterator.next();
  16.         //获取pname
  17.         String pname = key.getPname();
  18.         //将剩余的所有数据的pname进行替换
  19.         while(iterator.hasNext()){
  20.             iterator.next();
  21.             //替换pname
  22.             key.setPname(pname);
  23.             //将k,v写出去
  24.             context.write(key,NullWritable.get());
  25.         }
  26.     }
  27. }

RJDriver.java

  1. package com.atguigu.mr.reducejoin;
  3. import org.apache.hadoop.fs.Path;
  4. import org.apache.hadoop.io.NullWritable;
  5. import org.apache.hadoop.mapreduce.Job;
  6. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  7. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  9. public class RJDriver {
  10.     public static void main(String[] args) throws Exception {
  11.         Job job = Job.getInstance();
  12.         //设置自定义分组类
  13.         job.setGroupingComparatorClass(MyCompartor.class);
  14.         job.setMapperClass(RJMapper.class);
  15.         job.setReducerClass(RJReducer.class);
  16.         job.setMapOutputKeyClass(OrderBean.class);
  17.         job.setMapOutputValueClass(NullWritable.class);
  18.         job.setOutputKeyClass(OrderBean.class);
  19.         job.setOutputValueClass(NullWritable.class);
  20.         FileInputFormat.setInputPaths(job,new Path("C:\\io\\input9"));
  21.         FileOutputFormat.setOutputPath(job,new Path("C:\\io\\output9"));
  22.         job.waitForCompletion(true);
  23.     }
  24. }

3.MapJoin工作原理

  1. # 使用场景
  2. - Map join使用于一张表十分小,一张表很大的场景
  3. - 优点: 在Map端缓存多张表,提前处理业务逻辑,这样增加Map端业务,减少Reduce端数据的压力,尽可能的减少数据倾斜
  4. - 具体方法: 采用DistributedCache
  5. 1. 在Mapper的Setup阶段,将文件读取到缓存集合中
  6. 2. 在驱动函数中加载缓存
  7. // 缓存普通文件到Task运行节点
  8. job.addCacheFile(New URL("file://e:/cache/pd.txt");

4. 案例实操

需求:将商品信息表根据商品pid合并到订单数据表中

$07[Hadoop-MapReduce框架原理(下)] - 图10

OrderBean.java

  1. package com.atguigu.mr.mapjoin;
  3. import org.apache.hadoop.io.WritableComparable;
  5. import java.io.DataInput;
  6. import java.io.DataOutput;
  7. import java.io.IOException;
  9. public class OrderBean implements WritableComparable<OrderBean> {
  10.     private String id;
  11.     private String pid;
  12.     private String amount;
  13.     private String pname;
  15.     public OrderBean(){
  17.     }
  18.     public OrderBean(String id, String pid, String amount, String pname){
  19.         this.id = id;
  20.         this.pid = pid;
  21.         this.amount = amount;
  22.         this.pname = pname;
  24.     }
  26.     public String getId() {
  27.         return id;
  28.     }
  30.     public void setId(String id) {
  31.         this.id = id;
  32.     }
  34.     public String getPid() {
  35.         return pid;
  36.     }
  38.     public void setPid(String pid) {
  39.         this.pid = pid;
  40.     }
  42.     public String getAmount() {
  43.         return amount;
  44.     }
  46.     public void setAmount(String amount) {
  47.         this.amount = amount;
  48.     }
  50.     public String getPname() {
  51.         return pname;
  52.     }
  54.     public void setPname(String pname) {
  55.         this.pname = pname;
  56.     }
  57.     /** 
  58.     * @Description: 排序,先按照pid进行排序,pid相同再按照pname记性排序
  59.     * @Param: [o] 
  60.     * @return: int 
  61.     * @Author: jcsune
  62.     * @Date: 2021/7/15 
  63.     */ 
  64.     @Override
  65.     public int compareTo(OrderBean o) {
  66.         int pidValue = this.pid.compareTo(o.pid);
  67.         if(pidValue==0){//说明pid相同,在按照pname排序
  68.             return -this.pname.compareTo(o.pname);
  69.         }
  70.         return pidValue;
  71.     }
  72.     /**
  73.     * @Description: 序列化
  74.     * @Param: [dataOutput]
  75.     * @return: void
  76.     * @Author: jcsune
  77.     * @Date: 2021/7/15
  78.     */
  79.     @Override
  80.     public void write(DataOutput dataOutput) throws IOException {
  81.               dataOutput.writeUTF(id);
  82.               dataOutput.writeUTF(pid);
  83.               dataOutput.writeUTF(pname);
  84.               dataOutput.writeUTF(amount);
  85.     }
  86.     /**
  87.     * @Description: 反序列化
  88.     * @Param: [dataInput]
  89.     * @return: void
  90.     * @Author: jcsune
  91.     * @Date: 2021/7/15
  92.     */
  93.     @Override
  94.     public void readFields(DataInput dataInput) throws IOException {
  95.             id = dataInput.readUTF();
  96.             pid = dataInput.readUTF();
  97.             pname = dataInput.readUTF();
  98.             amount = dataInput.readUTF();
  99.     }
  101.     @Override
  102.     public String toString() {
  103.         return id + " " + pid + " " + amount + " " + pname;
  104.     }
  105. }

MJMapper.java

  1. package com.atguigu.mr.mapjoin;
  3. import org.apache.hadoop.fs.FSDataInputStream;
  4. import org.apache.hadoop.fs.FileSystem;
  5. import org.apache.hadoop.fs.Path;
  6. import org.apache.hadoop.io.LongWritable;
  7. import org.apache.hadoop.io.NullWritable;
  8. import org.apache.hadoop.io.Text;
  9. import org.apache.hadoop.mapreduce.Mapper;
  11. import java.io.BufferedReader;
  12. import java.io.IOException;
  13. import java.io.InputStreamReader;
  15. import java.net.URI;
  16. import java.util.HashMap;
  17. import java.util.Map;
  19. public class MJMapper extends Mapper<LongWritable, Text,OrderBean, NullWritable> {
  20.     private Map<String,String> map = new HashMap<String,String>();
  22.     /*
  23.     在任务开始的时候,只执行一次在map方法之前执行
  24.     作用 将pd.txt中的内容缓存起来
  25.      */
  26.     @Override
  27.     protected void setup(Context context) throws IOException, InterruptedException {
  28.         FileSystem fs = null;
  29.         BufferedReader br =null;
  30.         try {
  31.             //创建流-读取pd.txt
  32.              fs= FileSystem.get(context.getConfiguration());
  33.             //获取缓存文件
  34.             URI[] cacheFiles = context.getCacheFiles();
  35.             //创建流
  36.             FSDataInputStream fis = fs.open(new Path(cacheFiles[0]));
  37.             //读数据-- 一行一行的读
  38.             //字符缓冲流----将字节流转成字符流再套用字符缓冲流
  39.              br = new BufferedReader(new InputStreamReader(fis,"utf-8"));
  40.             //读数据
  41.             String line = "";
  42.             while((line = br.readLine())!= null){
  43.                 //将数据切割
  44.                 String[] split = line.split("\t");
  45.                 //将数据存放到map中
  46.                 map.put(split[0],split[1]);
  47.             }
  49.         } catch (IOException e) {
  50.             e.printStackTrace();
  51.         }finally {
  52.             if (br != null){
  53.                 br.close();
  54.             }
  55.             if(fs != null){
  56.                 fs.close();
  57.             }
  58.         }
  61.     }
  63.     @Override
  64.     protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
  65.         String line = value.toString();
  67.         //切割数据
  68.         String[] info = line.split("\t");
  69.         //封装对象
  70.         OrderBean orderBean = new OrderBean(info[0],info[1],info[2],map.get(info[1]));
  71.         //写出k,v
  72.         context.write(orderBean,NullWritable.get());
  73.     }
  74. }

MJDriver.class

  1. package com.atguigu.mr.mapjoin;
  3. import org.apache.hadoop.conf.Configuration;
  4. import org.apache.hadoop.fs.Path;
  5. import org.apache.hadoop.io.NullWritable;
  6. import org.apache.hadoop.mapreduce.Job;
  7. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  8. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  10. import java.net.URI;
  12. public class MJDriver {
  13.     public static void main(String[] args) throws Exception {
  14.         Job job = Job.getInstance(new Configuration());
  15.         job.addCacheFile(new URI("file:///C:/io/input10/pd.txt"));//添加缓存文件的路径,一定要注意斜杠
  16.         job.setMapperClass(MJMapper.class);
  17.         job.setNumReduceTasks(0);//表示没有ReducerTask
  18.         job.setMapOutputKeyClass(OrderBean.class);
  19.         job.setMapOutputValueClass(NullWritable.class);
  20.         job.setMapOutputKeyClass(OrderBean.class);
  21.         job.setMapOutputValueClass(NullWritable.class);
  22.         FileInputFormat.setInputPaths(job,new Path("C:\\io\\input10\\order.txt"));
  23.         FileOutputFormat.setOutputPath(job,new Path("C:\\io\\output10"));
  24.         job.waitForCompletion(true);
  28.     }
  29. }