Hadoop系列之-5、MapReduce高阶部分

素颜马尾好姑娘i 2022-09-03 00:21 120阅读 0赞

**[大数据系列文章目录][Link 1]**

### 目录 ###

*  深入MapReduce
 *   *  MapReduce的输入和输出
     *  MapReduce的处理流程解析
     *  Mapper任务执行过程详解
     *  Reducer任务执行过程详解
 *  MapReduce分区
 *   *  分区概述
     *  分区步骤
 *  MapReduce的排序和序列化
 *   *  概述
     *  需求
     *  分析
     *  实现
     *  自定义类型和比较器
     *  编写Mapper代码
     *  编写Reducer代码
     *  编写主类代码
 *  MapReuce的Combineer
 *   *  概念
     *  实现步骤
 *  MapReduce的自定义分组
 *   *  需求
     *  分析
     *  实现
     *   *  第一步：定义OrderBean
         *  第二步：自定义分区
         *  第三步：自定义groupingComparator
         *  第四步：程序main函数入口
 *  MapReduce的运行机制详解
 *   *  MapTask工作机制
     *   *  详细步骤
         *  配置
     *  ReduceTask工作机制
     *  MapReduce的shuffle过程
 *  MapReduce高阶训练
 *   *  上网流量统计
     *   *  需求：统计求和
         *   *  思路分析
             *  代码实现
 *  结束

> 上篇文章讲解了MapReduce的基础介绍和使用，这篇文章我准备来讲解MapReduce的原理和高阶部分，开搞。

# 深入MapReduce #

## MapReduce的输入和输出 ##

MapReduce框架运转在<key,value>键值对 上，也就是说，框架把作业的输入看成是一组<key,value>键值对，同样也产生一组<key,value>键值对作为作业的输出，这两组键值对可能是不同的。

一个MapReduce作业的输入和输出类型如下图所示：可以看出在整个标准的流程中，会有三组<key,value>键值对类型的存在。  
![在这里插入图片描述][f9c0677b96f540698edfead8bb9a0b62.png]

## MapReduce的处理流程解析 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2w4NDgxNjg_size_16_color_FFFFFF_t_70]

## Mapper任务执行过程详解 ##

第一阶段是把输入目录下文件按照一定的标准逐个进行逻辑切片，形成切片规划。默认情况下，Split size = Block size。每一个切片由一个MapTask处理。  
![在这里插入图片描述][30556aacd3c849c0bf94a896085d6db8.png]  
第二阶段是对切片中的数据按照一定的规则解析成<key,value>对。默认规则是把每一行文本内容解析成键值对。key是每一行的起始位置(单位是字节)，value是本行的文本内容。（TextInputFormat）  
![在这里插入图片描述][2253330215404c9ebef78e6d68363b9c.png]  
第三阶段是调用Mapper类中的map方法。上阶段中每解析出来的一个<k,v>，调用一次map方法。每次调用map方法会输出零个或多个键值对。

第四阶段是按照一定的规则对第三阶段输出的键值对进行分区。默认是只有一个区。分区的数量就是Reducer任务运行的数量。默认只有一个Reducer任务。

第五阶段是对每个分区中的键值对进行排序。首先，按照键进行排序，对于键相同的键值对，按照值进行排序。比如三个键值对<2,2>、<1,3>、<2,1>，键和值分别是整数。那么排序后的结果是<1,3>、<2,1>、<2,2>。如果有第六阶段，那么进入第六阶段；如果没有，直接输出到文件中。

第六阶段是对数据进行局部聚合处理，也就是combiner处理。键相等的键值对会调用一次reduce方法。经过这一阶段，数据量会减少。**本阶段默认是没有的。**

## Reducer任务执行过程详解 ##

第一阶段是Reducer任务会主动从Mapper任务复制其输出的键值对。Mapper任务可能会有很多，因此Reducer会复制多个Mapper的输出。

第二阶段是把复制到Reducer本地数据，全部进行合并，即把分散的数据合并成一个大的数据。再对合并后的数据排序。

第三阶段是对排序后的键值对调用reduce方法。键相等的键值对调用一次reduce方法，每次调用会产生零个或者多个键值对。最后把这些输出的键值对写入到HDFS文件中。

**在整个MapReduce程序的开发过程中，我们最大的工作量是覆盖map方法和覆盖reduce方法。**

# MapReduce分区 #

## 分区概述 ##

在 MapReduce 中, 通过我们指定分区, 会将同一个分区的数据发送到同一个Reduce当中进行处理。例如: 为了数据的统计, 可以把一批类似的数据发送到同一个 Reduce 当中, 在同一个 Reduce 当中统计相同类型的数据, 就可以实现类似的数据分区和统计等

其实就是相同类型的数据, 有共性的数据, 送到一起去处理, 在Reduce过程中，可以根据实际需求（比如按某个维度进行归档，类似于数据库的分组），把Map完的数据Reduce到不同的文件中。分区的设置需要与ReduceTaskNum配合使用。比如想要得到5个分区的数据结果。那么就得设置5个ReduceTask。

**需求：将以下数据进行分开处理**

详细数据参见partition.csv 这个文本文件，其中第五个字段表示开奖结果数值，现在需求将15以上的结果以及15以下的结果进行分开成两个文件进行保存  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2w4NDgxNjg_size_16_color_FFFFFF_t_70 1]

## 分区步骤 ##

**1、 定义 Mapper**  
这个 Mapper 程序不做任何逻辑, 也不对 Key-Value 做任何改变, 只是接收数据, 然后往下发送

public class MyMapper extends Mapper<LongWritable,Text,Text,NullWritable>{ 
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { 
            context.write(value,NullWritable.get());
        }
    }

**2.、自定义Partitioner**

主要的逻辑就在这里, 这也是这个案例的意义, 通过 Partitioner 将数据分发给不同的 Reducer

/** * 这里的输入类型与我们map阶段的输出类型相同 */
    public class MyPartitioner extends Partitioner<Text,NullWritable>{ 
        /** * 返回值表示我们的数据要去到哪个分区 * 返回值只是一个分区的标记，标记所有相同的数据去到指定的分区 */
        @Override
        public int getPartition(Text text, NullWritable nullWritable, int i) { 
            String result = text.toString().split("\t")[5];
            if (Integer.parseInt(result) > 15){ 
                return 1;
            }else{ 
                return 0;
            }
        }
    }

**3、定义 Reducer 逻辑**

这个 Reducer 也不做任何处理, 将数据原封不动的输出即可

public class MyReducer extends Reducer<Text,NullWritable,Text,NullWritable> { 
        @Override
        protected void reduce(Text key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException { 
            context.write(key,NullWritable.get());
        }
    }

**4、主类中设置分区类和ReduceTask个数**

public class PartitionerRunner { 
        public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { 
            //1、创建建一个job任务对象
            Configuration configuration = new Configuration();
            Job job = Job.getInstance(configuration, "mypartitioner");
    
            //2、指定job所在的jar包
            job.setJarByClass(PartitionerRunner.class);
    
            //3、指定源文件的读取方式类和源文件的读取路径
            job.setInputFormatClass(TextInputFormat.class); //按照行读取
            TextInputFormat.addInputPath(job, new Path("hdfs://node1:8020/input/partitioner")); //只需要指定源文件所在的目录即可
            // TextInputFormat.addInputPath(job, new Path("file:///E:\\input\\partitioner")); //只需要指定源文件所在的目录即可
    
            //4、指定自定义的Mapper类和K2、V2类型
            job.setMapperClass(PartitionerMapper.class); //指定Mapper类
            job.setMapOutputKeyClass(Text.class); //K2类型
            job.setMapOutputValueClass(NullWritable.class);//V2类型
    
            //5、指定自定义分区类（如果有的话）
            job.setPartitionerClass(MyPartitioner.class);
            //6、指定自定义分组类（如果有的话）
            //7、指定自定义的Reducer类和K3、V3的数据类型
            job.setReducerClass(PartitionerReducer.class); //指定Reducer类
            job.setOutputKeyClass(Text.class); //K3类型
            job.setOutputValueClass(NullWritable.class);  //V3类型
    
    
            //设置Reduce的个数
            job.setNumReduceTasks(2);
    
            //8、指定输出方式类和结果输出路径
            job.setOutputFormatClass(TextOutputFormat.class);
            TextOutputFormat.setOutputPath(job, new  Path("hdfs://node1:8020/output/partitioner")); //目标目录不能存在，否则报错
            //TextOutputFormat.setOutputPath(job, new Path("file:///E:\\output\\partitoner")); //目标目录不能存在，否则报错
    
            //9、将job提交到yarn集群
            boolean bl = job.waitForCompletion(true); //true表示可以看到任务的执行进度
    
            //10.退出执行进程
            System.exit(bl?0:1);
        }
    }

# MapReduce的排序和序列化 #

## 概述 ##

**序列化（Serialization）**是指把结构化对象转化为字节流。  
**反序列化（Serialization）**（Deserialization）是序列化的逆过程。把字节流转为结构化对象。

当要在进程间传递对象或持久化对象的时候，就需要序列化对象成字节流，反之当要将接收到或从磁盘读取的字节流转换为对象，就要进行反序列化。

Java的序列化（Serializable）是一个重量级序列化框架，一个对象被序列化后，会附带很多额外的信息（各种校验信息，header，继承体系…），不便于在网络中高效传输；所以，hadoop自己开发了一套序列化机制（Writable），精简，高效。不用像java对象类一样传输多层的父子关系，需要哪个属性就传输哪个属性值，大大的减少网络传输的开销。

Writable是Hadoop的序列化格式，hadoop定义了这样一个Writable接口。  
一个类要支持可序列化只需实现这个接口即可。

public interface  Writable { 
     void write(DataOutput out) throws IOException;
     void readFields(DataInput in) throws IOException;
    }

另外 Writable 有一个子接口是 WritableComparable, WritableComparable 是既可实现序列化, 也可以对key进行比较, 我们这里可以通过自定义 Key 实现 WritableComparable 来实现我们的排序功能.

// WritableComparable分别继承Writable和Comparable
    public interface WritableComparable<T> extends Writable, Comparable<T> { 
    }
    //Comparable
    public interface Comparable<T> { 
        int compareTo(T var1);
    }

Comparable接口中的comparaTo方法用来定义排序规则，用于将当前对象与方法的参数进行比较。  
例如：o1.compareTo(o2);

**如果指定的数与参数相等返回0。  
如果指定的数小于参数返回 -1。  
如果指定的数大于参数返回 1。**

返回正数的话，当前对象（调用compareTo方法的对象o1）要排在比较对象（compareTo传参对象o2）后面，返回负数的话，放在前面。

## 需求 ##

数据格式如下

a   1
      a   9
      b   3
      a   7
      b   8
      b   10
      a   5

要求:  
第一列按照字典顺序进行排列

第一列相同的时候, 第二列按照升序进行排列

## 分析 ##

**实现自定义的bean来封装数据，并将bean作为map输出的key来传输**

MR程序在处理数据的过程中会对数据排序(map输出的kv对传输到reduce之前，会排序)，排序的依据是map输出的key。所以，我们如果要实现自己需要的排序规则，则可以考虑将排序因素放到key中，让key实现接口：WritableComparable，然后重写key的compareTo方法。

如果自定义的JavaBean要参与MapReduce运算，则必须进行序列化，必须实现 Writable接口，如果该JavaBean作为K2，则必须实现WritableComparable接口，让JavaBean具有排序的功能

## 实现 ##

## 自定义类型和比较器 ##

public class SortBean implements WritableComparable<SortBean>{ 
    
      private String word;
      private int  num;
    
      public String getWord() { 
    	  return word;
      }
    
      public void setWord(String word) { 
    	  this.word = word;
      }
    
      public int getNum() { 
    	  return num;
      }
    
      public void setNum(int num) { 
    	  this.num = num;
      }
    
      @Override
      public String toString() { 
    	  return   word + "\t"+ num ;
      }
    
      //实现比较器，指定排序的规则
      /* 规则: 第一列(word)按照字典顺序进行排列 // aac aad 第一列相同的时候, 第二列(num)按照升序进行排列 */
      /* a 1 a 5 b 3 b 8 */
      @Override
      public int compareTo(SortBean sortBean) { 
    	  //先对第一列排序: Word排序
    	  int result = this.word.compareTo(sortBean.word);
    	  //如果第一列相同，则按照第二列进行排序
    	  if(result == 0){ 
    		  return  this.num - sortBean.num;
    	  }
    	  return result;
      }
    
      //实现序列化
      @Override
      public void write(DataOutput out) throws IOException { 
    	  out.writeUTF(word);
    	  out.writeInt(num);
      }
    
      //实现反序列
      @Override
      public void readFields(DataInput in) throws IOException { 
    		  this.word = in.readUTF();
    		  this.num = in.readInt();
      }
    }

## 编写Mapper代码 ##

public class SortMapper extends Mapper<LongWritable,Text,SortBean,NullWritable> { 
      /* map方法将K1和V1转为K2和V2: K1 V1 0 a 3 5 b 7 ---------------------- K2 V2 SortBean(a 3) NullWritable SortBean(b 7) NullWritable */
      @Override
      protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { 
    	  //1:将行文本数据(V1)拆分，并将数据封装到SortBean对象,就可以得到K2
    	  String[] split = value.toString().split("\t");
    
    	  SortBean sortBean = new SortBean();
    	  sortBean.setWord(split[0]);
    	  sortBean.setNum(Integer.parseInt(split[1]));
    
    	  //2:将K2和V2写入上下文中
    	  context.write(sortBean, NullWritable.get());
      }
     }

## 编写Reducer代码 ##

public class SortReducer extends Reducer<SortBean,NullWritable,SortBean,NullWritable> { 
     
       //reduce方法将新的K2和V2转为K3和V3
       @Override
       protected void reduce(SortBean key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException { 
          context.write(key, NullWritable.get());
       }
    }

## 编写主类代码 ##

public class SortRunner { 
          public static void main(String[] args) throws Exception { 
    		  
    		  Configuration conf = new Configuration();
              //1:创建job对象
              Job job = Job.getInstance(conf, "mapreduce_sort");
       
           
    	       //2:指定job所在的jar包
    		  job.setJarByClass(SortRunner.class);
        
    		  //3:指定源文件的读取方式类和源文件的读取路径
    		  job.setInputFormatClass(TextInputFormat.class);
    		  ///TextInputFormat.addInputPath(job, new Path("hdfs://node01:8020/input/sort_input"));
    		  TextInputFormat.addInputPath(job, new Path("file:///D:\\input\\sort_input"));
    
    		  //4:指定自定义的Mapper类和K2、V2类型
    		  job.setMapperClass(SortMapper.class);
    		  job.setMapOutputKeyClass(SortBean.class);
    		  job.setMapOutputValueClass(NullWritable.class);
    
    
    		  //5:指定自定义的Reducer类和K3、V3的数据类型
    		  job.setReducerClass(SortReducer.class);
    		  job.setOutputKeyClass(SortBean.class);
    		  job.setOutputValueClass(NullWritable.class);
    
    
    		  //6:指定输出方式类和结果输出路径
    		  job.setOutputFormatClass(TextOutputFormat.class);
    		  TextOutputFormat.setOutputPath(job, new Path("file:///D:\\out\\sort_out"));
    
       
             //7:将job提交给yarn集群
             boolean bl = job.waitForCompletion(true);
    
             System.exit(bl?0:1);
          }
      }

# MapReuce的Combineer #

## 概念 ##

每一个 map 都可能会产生大量的本地输出，Combiner 的作用就是对 map 端的输出先做一次合并，以减少在 map 和 reduce 节点之间的数据传输量，以提高网络IO 性能，是 MapReduce 的一种优化手段之一  
combiner 是 MR 程序中 Mapper 和 Reducer 之外的一种组件  
combiner 组件的父类就是 Reducer  
combiner 和 reducer 的区别在于运行的位置  
　　Combiner 是在每一个 maptask 所在的节点运行  
　　Reducer 是接收全局所有 Mapper 的输出结果  
combiner 的意义就是对每一个 maptask 的输出进行局部汇总，以减小网络传输量

## 实现步骤 ##

在这里以单词统计为例,实现Combiner  
**1、**自定义一个 combiner 继承 Reducer，重写 reduce 方法

public class MyCombiner extends Reducer<Text,LongWritable,Text,LongWritable> { 
    
    
        /* key : hello values: <1,1,1,1> */
        @Override
        protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException { 
            long count = 0;
            //1:遍历集合，将集合中的数字相加，得到 V3
            for (LongWritable value : values) { 
                count += value.get();
            }
            //2:将K3和V3写入上下文中
            context.write(key, new LongWritable(count));
        }
    }

**2、**在 job 中设置 job.setCombinerClass(CustomCombiner.class)

job.setCombinerClass(MyCombiner.class);

combiner 能够应用的前提是不能影响最终的业务逻辑，而且，combiner 的输出 kv 应该跟 reducer 的输入 kv 类型要对应起来

**3、**对使用Combiner之前和之后的日志进行对比  
![在这里插入图片描述][0443b8c5ce2e412e96e4c3b372a2766f.png]  
通过对比发现，使用Combiner之后，Reduce输入的键值对数量降低了，提供了网络传输效率。

# MapReduce的自定义分组 #

GroupingComparator是mapreduce当中reduce端的一个功能组件，主要的作用是决定哪些数据作为一组，调用一次reduce的逻辑，默认是每个不同的key，作为多个不同的组，每个组调用一次reduce逻辑，我们可以自定义GroupingComparator实现不同的key作为同一个组，调用一次reduce逻辑

## 需求 ##

有如下订单数据  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2w4NDgxNjg_size_16_color_FFFFFF_t_70 2]  
现在需要求出每一个订单中成交金额最大的一笔交易

## 分析 ##

**1、**利用“订单id和成交金额”作为key，可以将map阶段读取到的所有订单数据按照id分区，按照金额排序，发送到reduce  
**2、**在reduce端利用groupingcomparator将订单id相同的kv聚合成组，然后取第一个即是最大值

## 实现 ##

### 第一步：定义OrderBean ###

定义一个OrderBean，里面定义两个字段，第一个字段是我们的orderId，第二个字段是我们的金额（注意金额一定要使用Double或者DoubleWritable类型，否则没法按照金额顺序排序）

public class OrderBean implements WritableComparable<OrderBean> { 
        private String orderId;
        private Double price;
        @Override
        public int compareTo(OrderBean o) { 
            //比较订单id的排序顺序
            int i = this.orderId.compareTo(o.orderId);
            if(i==0){ 
              //如果订单id相同，则比较金额，金额大的排在前面
               i = - this.price.compareTo(o.price);
            }
            return i;
        }
        @Override
        public void write(DataOutput out) throws IOException { 
                out.writeUTF(orderId);
                out.writeDouble(price);
        }
        @Override
        public void readFields(DataInput in) throws IOException { 
            this.orderId =  in.readUTF();
            this.price = in.readDouble();
        }
        public OrderBean() { 
        }
        public OrderBean(String orderId, Double price) { 
            this.orderId = orderId;
            this.price = price;
        }
        public String getOrderId() { 
            return orderId;
        }
        public void setOrderId(String orderId) { 
            this.orderId = orderId;
        }
        public Double getPrice() { 
            return price;
        }
        public void setPrice(Double price) { 
            this.price = price;
        }
        @Override
        public String toString() { 
            return  orderId +"\t"+price;
        }
    }

### 第二步：自定义分区 ###

自定义分区，按照订单id进行分区，把所有订单id相同的数据，都发送到同一个reduce中去

public class OrderPartition extends Partitioner<OrderBean,NullWritable> { 
        @Override
        public int getPartition(OrderBean orderBean, NullWritable nullWritable, int i) { 
            //自定义分区，将相同订单id的数据发送到同一个reduce里面去
            return  (orderBean.getOrderId().hashCode() & Integer.MAX_VALUE)%i;
        }
    }

### 第三步：自定义groupingComparator ###

按照我们自己的逻辑进行分组，通过比较相同的订单id，将相同的订单id放到一个组里面去，进过分组之后当中的数据，已经全部是排好序的数据，我们只需要取前topN即可

/* 1: 继承WriteableComparator 2: 调用父类的有参构造 3: 指定分组的规则(重写方法) */
    
    import org.apache.hadoop.io.WritableComparable;
    import org.apache.hadoop.io.WritableComparator;
    
    // 1: 继承WriteableComparator
    public class OrderGroupComparator extends WritableComparator { 
        // 2: 调用父类的有参构造
        public OrderGroupComparator() { 
            super(OrderBean.class,true);
        }
    
        //3: 指定分组的规则(重写方法)
        @Override
        public int compare(WritableComparable a, WritableComparable b) { 
            //3.1 对形参做强制类型转换
            OrderBean first = (OrderBean)a;
            OrderBean second = (OrderBean)b;
    
            //3.2 指定分组规则
            return first.getOrderId().compareTo(second.getOrderId());
        }
    }

### 第四步：程序main函数入口 ###

public class GroupingRunner { 
        public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { 
            //1、创建建一个job任务对象
            Configuration configuration = new Configuration();
            Job job = Job.getInstance(configuration, "grouping_demo");
    
            //2、指定job所在的jar包
            job.setJarByClass(GroupingRunner.class);
    
            //3、指定源文件的读取方式类和源文件的读取路径
            job.setInputFormatClass(TextInputFormat.class); //按照行读取
            //TextInputFormat.addInputPath(job, new Path("hdfs://node1:8020/input/wordcount")); //只需要指定源文件所在的目录即可
            TextInputFormat.addInputPath(job, new Path("file:///E:\\input\\grouping_demo")); //只需要指定源文件所在的目录即可
    
            //4、指定自定义的Mapper类和K2、V2类型
            job.setMapperClass(GroupingMapper.class); //指定Mapper类
            job.setMapOutputKeyClass(OrderBean.class); //K2类型
            job.setMapOutputValueClass(Text.class);//V2类型
    
            //5、指定自定义分区类（如果有的话）
            job.setPartitionerClass(MyPartitioner.class);
            //6、指定自定义分组类（如果有的话）
            job.setGroupingComparatorClass(GroupingComparator.class);
            //7、指定自定义Combiner类(如果有的话)
            //job.setCombinerClass(MyCombiner.class);
    
    
            //设置ReduceTask个数
            job.setNumReduceTasks(3);
    
            //8、指定自定义的Reducer类和K3、V3的数据类型
            job.setReducerClass(GroupingReducer.class); //指定Reducer类
            job.setOutputKeyClass(Text.class); //K3类型
            job.setOutputValueClass(NullWritable.class);  //V3类型
    
            //9、指定输出方式类和结果输出路径
            job.setOutputFormatClass(TextOutputFormat.class);
            //TextOutputFormat.setOutputPath(job, new Path("hdfs://node1:8020/output/wordcount")); //目标目录不能存在，否则报错
            TextOutputFormat.setOutputPath(job, new  Path("file:///E:\\output\\grouping_demo")); //目标目录不能存在，否则报错
    
            //10、将job提交到yarn集群
            boolean bl = job.waitForCompletion(true); //true表示可以看到任务的执行进度
    
            //11.退出执行进程
            System.exit(bl?0:1);
        }
    }

# MapReduce的运行机制详解 #

![在这里插入图片描述][96d12f417c9640cf8b3f6a4aa85b6fb0.png]

## MapTask工作机制 ##

**简单概述：**inputFile通过split被逻辑切分为多个split文件，通过Record按行读取内容给map（用户自己实现的）进行处理，数据被map处理结束之后交给OutputCollector收集器，对其结果key进行分区（默认使用hash分区），然后写入buffer，每个map task都有一个内存缓冲区，存储着map的输出结果，当缓冲区快满的时候需要将缓冲区的数据以一个临时文件的方式存放到磁盘，当整个map task结束后再对磁盘中这个map task产生的所有临时文件做合并，生成最终的正式输出文件，然后等待reduce task来拉数据。

### 详细步骤 ###

**1、** 首先，**读取数据组件InputFormat**（默认TextInputFormat）会通过getSplits方法对输入目录中文件进行逻辑切片规划得到splits，有多少个split就对应启动多少个MapTask。split与block的对应关系默认是一对一。

**2、** 将输入文件切分为splits之后，由**RecordReader对象（默认LineRecordReader）进行读取**，以\\n作为分隔符，读取一行数据，返回<key，value>。Key表示每行首字符偏移值，value表示这一行文本内容。

**3、** 读取split返回<key,value>，进入用户自己继承的Mapper类中，**执行用户重写的map函数**。RecordReader读取一行这里调用一次。

**4、** map逻辑完之后，将map的每条结果通过context.write进行collect数据收集。在collect中，会先对其进行分区处理，默认使用HashPartitioner。

**MapReduce 提供 Partitioner 接口, 它的作用就是根据 Key 或 Value 及 Reducer 的数量来决定当前的这对输出数据最终应该交由哪个 Reduce task 处理, 默认对 Key Hash 后再以 Reducer 数量取模. 默认的取模方式只是为了平均 Reducer 的处理能力, 如果用户自己对 Partitioner 有需求, 可以订制并设置到 Job 上**  
**5、** 接下来, 会将数据写入内存, 内存中这片区域叫做环形缓冲区, 缓冲区的作用是批量收集 Mapper 结果, 减少磁盘IO的影响. 我们的 Key/Value 对以及 Partition 的结果都会被写入缓冲区. 当然, 写入之前，Key 与 Value 值都会被序列化成字节数组

**环形缓冲区其实是一个数组, 数组中存放着 Key, Value 的序列化数据和 Key, Value 的元数据信息, 包括 Partition, Key 的起始位置, Value 的起始位置以及 Value 的长度. 环形结构是一个抽象概念  
缓冲区是有大小限制, 默认是 100MB. 当 Mapper 的输出结果很多时, 就可能会撑爆内存, 所以需要在一定条件下将缓冲区中的数据临时写入磁盘, 然后重新利用这块缓冲区. 这个从内存往磁盘写数据的过程被称为 Spill, 中文可译为溢写. 这个溢写是由单独线程来完成, 不影响往缓冲区写 Mapper 结果的线程. 溢写线程启动时不应该阻止 Mapper 的结果输出, 所以整个缓冲区有个溢写的比例 spill.percent. 这个比例默认是 0.8, 也就是当缓冲区的数据已经达到阈值 buffer size \* spill percent = 100MB \* 0.8 = 80MB, 溢写线程启动, 锁定这 80MB 的内存, 执行溢写过程. Mapper 的输出结果还可以往剩下的 20MB 内存中写, 互不影响**

**6、** 当溢写线程启动后, 需要对这 80MB 空间内的 Key 做排序 (Sort). 排序是 MapReduce 模型默认的行为, 这里的排序也是对序列化的字节做的排序  
**如果 Job 设置过 Combiner, 那么现在就是使用 Combiner 的时候了. 将有相同 Key 的 Key/Value 对的 Value 加起来, 减少溢写到磁盘的数据量. Combiner 会优化 MapReduce 的中间结果, 所以它在整个模型中会多次使用**

**7、** 合并溢写文件, 每次溢写会在磁盘上生成一个临时文件 (写之前判断是否有 Combiner), 如果 Mapper 的输出结果真的很大, 有多次这样的溢写发生, 磁盘上相应的就会有多个临时文件存在. 当整个数据处理结束之后开始对磁盘中的临时文件进行 Merge 合并, 因为最终的文件只有一个, 写入磁盘, 并且为这个文件提供了一个索引文件, 以记录每个reduce对应数据的偏移量

mapred-site.xml

### 配置 ###

<table> 
 <thead> 
  <tr> 
   <th>配置</th> 
   <th>默认值</th> 
   <th>解释</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>mapreduce.task.io.sort.mb</td> 
   <td>100</td> 
   <td>设置环型缓冲区的内存值大小</td> 
  </tr> 
  <tr> 
   <td>mapreduce.map.sort.spill.percent</td> 
   <td>0.8</td> 
   <td>设置溢写的比例</td> 
  </tr> 
  <tr> 
   <td>mapreduce.cluster.local.dir</td> 
   <td>${hadoop.tmp.dir}/mapred/local</td> 
   <td>溢写数据目录</td> 
  </tr> 
  <tr> 
   <td>mapreduce.task.io.sort.factor</td> 
   <td>10</td> 
   <td>设置一次合并多少个溢写文件</td> 
  </tr> 
 </tbody> 
</table>

## ReduceTask工作机制 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2w4NDgxNjg_size_16_color_FFFFFF_t_70 3]  
Reduce大致分为copy、sort、reduce三个阶段，重点在前两个阶段。copy阶段包含一个eventFetcher来获取已完成的map列表，由Fetcher线程去copy数据，在此过程中会启动两个merge线程，分别为inMemoryMerger和onDiskMerger，分别将内存中的数据merge到磁盘和将磁盘中的数据进行merge。待数据copy完成之后，copy阶段就完成了，开始进行sort阶段，sort阶段主要是执行finalMerge操作，纯粹的sort阶段，完成之后就是reduce阶段，调用用户定义的reduce函数进行处理。

**详细步骤**：  
**1、**  Copy阶段，简单地拉取数据。Reduce进程启动一些数据copy线程(Fetcher)，通过HTTP方式请求maptask获取属于自己的文件。

**2、** Merge阶段 ，这里的merge如map端的merge动作，只是数组中存放的是不同map端copy来的数值。Copy过来的数据会先放入内存缓冲区中，这里的缓冲区大小要比map端的更为灵活。merge有三种形式：内存到内存；内存到磁盘；磁盘到磁盘。默认情况下第一种形式不启用。当内存中的数据量到达一定阈值，就启动内存到磁盘的merge。与map 端类似，这也是溢写的过程，然后在磁盘中生成了众多的溢写文件。第二种merge方式一直在运行，直到没有map端的数据时才结束，然后启动第三种磁盘到磁盘的merge方式生成最终的文件。

**3、** 合并排序 ，把分散的数据合并成一个大的数据后，还会再对合并后的数据排序。

**4、** 对排序后的键值对调用reduce方法 ，键相等的键值对调用一次reduce方法，每次调用会产生零个或者多个键值对，最后把这些输出的键值对写入到HDFS文件中。

## MapReduce的shuffle过程 ##

map阶段处理的数据如何传递给reduce阶段，是MapReduce框架中最关键的一个流程，这个流程就叫shuffle。  
shuffle: 洗牌、发牌——（核心机制：数据分区，排序，Combiner，分组等过程）。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2w4NDgxNjg_size_16_color_FFFFFF_t_70 4]  
shuffle是Mapreduce的核心，它分布在Mapreduce的map阶段和reduce阶段。一般把从Map产生输出开始到Reduce取得数据作为输入之前的过程称作shuffle。

**1）Collect阶段**：将MapTask的结果输出到默认大小为100M的环形缓冲区，保存的是key/value，Partition分区信息等。  
**2）Spill阶段**：当内存中的数据量达到一定的阀值的时候，就会将数据写入本地磁盘，在将数据写入磁盘之前需要对数据进行一次排序的操作，如果配置了combiner，还会将有相同分区号和key的数据进行排序。  
**3）Merge阶段**：把所有溢出的临时文件进行一次合并操作，以确保一个MapTask最终只产生一个中间数据文件。  
**4）Copy阶段**：ReduceTask启动Fetcher线程到已经完成MapTask的节点上复制一份属于自己的数据，这些数据默认会保存在内存的缓冲区中，当内存的缓冲区达到一定的阀值的时候，就会将数据写到磁盘之上。  
**5）Merge阶段**：在ReduceTask远程复制数据的同时，会在后台开启两个线程对内存到本地的数据文件进行合并操作。  
**6）Sort阶段**：在对数据进行合并的同时，会进行排序操作，由于MapTask阶段已经对数据进行了局部的排序，ReduceTask只需保证Copy的数据的最终整体有效性即可。  
Shuffle中的缓冲区大小会影响到mapreduce程序的执行效率，原则上说，缓冲区越大，磁盘io的次数越少，执行速度就越快

缓冲区的大小可以通过参数调整, 参数：mapreduce.task.io.sort.mb 默认100M

# MapReduce高阶训练 #

## 上网流量统计 ##

数据格式如下:  
![在这里插入图片描述][4ebfa2536d254450bd7994a25750bc67.png]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2w4NDgxNjg_size_16_color_FFFFFF_t_70 5]

### 需求：统计求和 ###

统计每个手机号的上行数据包数总和，下行数据包数总和，上行总流量之和，下行总流量之和  
分析：以手机号码作为key值，上行数据包，下行数据包，上行总流量，下行总流量四个字段作为value值，然后以这个key和value作为map阶段的输出，reduce阶段的输入。

#### 思路分析 ####

![在这里插入图片描述][698a1800c1784de0b569f63b65083318.png]

#### 代码实现 ####

**第一步：自定义map的输出value对象FlowBean**

import org.apache.hadoop.io.Writable;
    import org.apache.hadoop.io.WritableComparable;
    
    import java.io.DataInput;
    import java.io.DataOutput;
    import java.io.IOException;
    //如果MR中的JavaBean不是作为K2，则只需要实现Writable接口即可
    public class FlowBean implements Writable{ 
        private Integer upFlow;    //上行包数
        private Integer  downFlow; //下行包数
        private Integer upCountFlow; //上行流量和
        private Integer downCountFlow; //下行流量和
    
    
        public FlowBean() { 
        }
    
        public FlowBean(Integer upFlow, Integer downFlow, Integer upCountFlow, Integer downCountFlow) { 
            this.upFlow = upFlow;
            this.downFlow = downFlow;
            this.upCountFlow = upCountFlow;
            this.downCountFlow = downCountFlow;
        }
    
        public Integer getUpFlow() { 
            return upFlow;
        }
    
        @Override
        public String toString() { 
            return upFlow +
                    "\t" + downFlow +
                    "\t" + upCountFlow +
                    "\t" + downCountFlow;
        }
    
        public void setUpFlow(Integer upFlow) { 
            this.upFlow = upFlow;
        }
    
        public Integer getDownFlow() { 
            return downFlow;
        }
    
        public void setDownFlow(Integer downFlow) { 
            this.downFlow = downFlow;
        }
    
        public Integer getUpCountFlow() { 
            return upCountFlow;
        }
    
        public void setUpCountFlow(Integer upCountFlow) { 
            this.upCountFlow = upCountFlow;
        }
    
        public Integer getDownCountFlow() { 
            return downCountFlow;
        }
    
        public void setDownCountFlow(Integer downCountFlow) { 
            this.downCountFlow = downCountFlow;
        }
    
        //序列化
        @Override
        public void write(DataOutput out) throws IOException { 
            out.writeInt(upFlow);
            out.writeInt(downFlow);
            out.writeInt(upCountFlow);
            out.writeInt(downCountFlow);
        }
        //反序列化
        @Override
        public void readFields(DataInput in) throws IOException { 
            this.upFlow = in.readInt();
            this.downFlow = in.readInt();
            this.upCountFlow = in.readInt();
            this.downCountFlow = in.readInt();
        }
    }

**第二步：定义FlowMapper类**

import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Mapper;
    
    import java.io.IOException;
    
    public class FlowCountMapper extends Mapper<LongWritable, Text,Text,FlowBean> { 
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { 
            //1:拆分行文本数据（拆分v1）
            String[] split = value.toString().split("\t");
    
            //2:从拆分数组中得到手机号，得到K2
            String phoneNum = split[1];
            //3:从拆分数组中得到4个流量字段，并封装到FlowBean,得到V2
            FlowBean flowBean = new FlowBean();
            flowBean.setUpFlow(Integer.parseInt(split[6]));
            flowBean.setDownFlow(Integer.parseInt(split[7]));
            flowBean.setUpCountFlow(Integer.parseInt(split[8]));
            flowBean.setDownCountFlow(Integer.parseInt(split[9]));
    
            //4:将K2和V2写入上下文中
            context.write(new Text(phoneNum), flowBean);
    
        }
    }

**第三步：定义FlowCountReducer类**

import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Reducer;
    
    import java.io.IOException;
    
    public class FlowCountReducer extends Reducer<Text,FlowBean,Text,FlowBean> { 
        @Override
        protected void reduce(Text key, Iterable<FlowBean> values, Context context) throws IOException, InterruptedException { 
            //1:定义四个变量，分别来存储上行包数，下行包数，上行流量总和，下行流量总和
            int upFlow = 0;
            int downFlow = 0;
            int upCountFlow = 0;
            int downCountFlow = 0;
    
            //2:遍历集合，将集合中每一个FlowBean的四个流量字段相加
            for (FlowBean flowBean : values) { 
                upFlow += flowBean.getUpFlow();
                downFlow += flowBean.getDownFlow();
                upCountFlow += flowBean.getUpCountFlow();
                downCountFlow += flowBean.getDownCountFlow();
            }
            //3:K3就是原来的K2，V3就是新的FlowBean
            FlowBean flowBean = new FlowBean(upFlow, downFlow, upCountFlow, downCountFlow);
    
            //4:将K3和V3写入上下文中
            context.write(key, flowBean);
    
        }
    }

**第四步：程序main函数入口FlowCountRunner**

import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.NullWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Job;
    import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
    
    import java.io.IOException;
    
    public class FlowCountRunner { 
        public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { 
            //1、创建建一个job任务对象
            Configuration configuration = new Configuration();
    
    
            Job job = Job.getInstance(configuration, "flowcount_demo");
    
            //2、指定job所在的jar包
            job.setJarByClass(FlowCountRunner.class);
    
            //3、指定源文件的读取方式类和源文件的读取路径
            job.setInputFormatClass(TextInputFormat.class); //按照行读取
            //TextInputFormat.addInputPath(job, new Path("hdfs://node1:8020/input/wordcount")); //只需要指定源文件所在的目录即可
            TextInputFormat.addInputPath(job, new Path("file:///E:\\input\\flowcount")); //只需要指定源文件所在的目录即可
    
            //4、指定自定义的Mapper类和K2、V2类型
            job.setMapperClass(FlowCountMapper.class); //指定Mapper类
            job.setMapOutputKeyClass(Text.class); //K2类型
            job.setMapOutputValueClass(FlowBean.class);//V2类型
    
            //5、指定自定义分区类（如果有的话）
            //6、指定自定义分组类（如果有的话）
            //7、指定自定义Combiner类(如果有的话)
    
    
            //设置ReduceTask个数
    
            //8、指定自定义的Reducer类和K3、V3的数据类型
            job.setReducerClass(FlowCountReducer.class); //指定Reducer类
            job.setOutputKeyClass(Text.class); //K3类型
            job.setOutputValueClass(FlowBean.class);  //V3类型
    
            //9、指定输出方式类和结果输出路径
            job.setOutputFormatClass(TextOutputFormat.class);
            //TextOutputFormat.setOutputPath(job, new Path("hdfs://node1:8020/output/wordcount")); //目标目录不能存在，否则报错
            TextOutputFormat.setOutputPath(job, new  Path("file:///E:\\output\\flowcount")); //目标目录不能存在，否则报错
    
            //10、将job提交到yarn集群
            boolean bl = job.waitForCompletion(true); //true表示可以看到任务的执行进度
    
            //11.退出执行进程
            System.exit(bl?0:1);
        }
    }

# 结束 #

MapReduce的文章共分为2篇，一篇基础一篇高阶，到这里MapReduce的教程就算结束，白嫖到这，我劝你要善良，点赞评论再走：）

[Link 1]: https://blog.csdn.net/l848168/article/details/119147380
[f9c0677b96f540698edfead8bb9a0b62.png]: /images/20220829/c0658ac2dfd44c8dace9954296834c5a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2w4NDgxNjg_size_16_color_FFFFFF_t_70]: /images/20220829/83665ed3f7c64675b283957333d14111.png
[30556aacd3c849c0bf94a896085d6db8.png]: /images/20220829/34d10ea51b5744b1ac090e4cea97b07d.png
[2253330215404c9ebef78e6d68363b9c.png]: /images/20220829/843fbceb67ad42bf99ff3982edc6e810.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2w4NDgxNjg_size_16_color_FFFFFF_t_70 1]: /images/20220829/abac4e7510154e6280a7eccc74287456.png
[0443b8c5ce2e412e96e4c3b372a2766f.png]: /images/20220829/be9b4dac0a43455e955c317363e29695.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2w4NDgxNjg_size_16_color_FFFFFF_t_70 2]: /images/20220829/6a301257ec9c4ef4b94e7f20bec4da84.png
[96d12f417c9640cf8b3f6a4aa85b6fb0.png]: /images/20220829/55f42ca655344771852f743db8c36484.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2w4NDgxNjg_size_16_color_FFFFFF_t_70 3]: /images/20220829/6e161245fabb4cca844ecc8cb2bc39a7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2w4NDgxNjg_size_16_color_FFFFFF_t_70 4]: /images/20220829/e8748928f03e44e9b70137dea6d91a87.png
[4ebfa2536d254450bd7994a25750bc67.png]: /images/20220829/86a9dcb45c184ac58c10b468ef6991eb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2w4NDgxNjg_size_16_color_FFFFFF_t_70 5]: /images/20220829/20af706e44ef4dea935a10f90ba2cf3c.png
[698a1800c1784de0b569f63b65083318.png]: /images/20220829/8341127b4c6443b789000fee90c0d3e6.png