第一章.InputFormat数据输入

$06[Hadoop-MapResource框架原理(上)] - 图1

1.切片与MapTask并行度决定机制

  1. # MapTesk并行度决定机制
  2. - 数据块:Block是HDFS物理上把数据分成一块一块
  3. - 数据切片只是在逻辑上对输入进行切片,并不会在磁盘上将其切成片进行存储

$06[Hadoop-MapResource框架原理(上)] - 图2

2.Job提交流程

$06[Hadoop-MapResource框架原理(上)] - 图3

  1. # FileInputFormat切片源码解析
  2. 1. 程序先找到你数据存储的目录
  3. 2. 开始遍历处理(规划切片)目录下的每一个文件
  4. 3. 遍历第一个文件ss.txt
  5.   a. 获取文件大小 fs.sizeOf(ss.txt)
  6.   b. 计算切片大小
  7.   c. 默认情况下,切片大小等于块的大小
  8.   d. 开始切,每次切片时,要判断切完剩下的部分是否大于块的1.1倍,不大于1.1倍就划分一块切片
  9.   e. 将切片信息写到一个切片规划文件中
  10.   f. 整个切片的核心过程在getSpilt()方法中完成
  11.   g. InputSpilt只记录了切片的元数据信息,比如起始位置,长度以及所在的节点列表等
  12. 4. 提交切片规划文件到YARN上,Yarn上的MrAPPMaster就可以根据切片规划文件计算开启MapTask个数

3.FileInputFormat切片机制

  1. # 切片机制
  2. 1. 简单的按照文件的内容长度进行切片
  3. 2. 切片大小默认等于块的大小
  4. 3. 切片时不考虑数据集整体而是逐个针对每一个文件单独切片

$06[Hadoop-MapResource框架原理(上)] - 图4

4.CombineTextInputFormat切片机制

  1. 框架默认的TextInputFormat切片机制是对任务按文件规划切片,不管文件多小,都会是一个单独的切片,都会交给一个MapTask,这样如果有大量小文件,就会产生大量的MapTask,处理效率极其低下
  3.  CombineTextInputFormat用于小文件过多的场景,它可以将多个小文件从逻辑上规划到一个切片中,这样多个小文件就可以交给一个MapTask处理
  1. # 虚拟存储切片最大值设置
  2. CombineTextInputFormat.setMaxInputSplitSize(job,4194304);//4M

切片机制

$06[Hadoop-MapResource框架原理(上)] - 图5

5.CombineTextInputFormat案例实操

  1. # 需求
  2. - 将输入的大量小文件合并成一个切片统一处理
  3. - 输入数据:准备四个小文件
  4. - 期望一个切片处理四个文件

实现过程:只需将驱动类中增加如下代码即可

  1. //设置虚拟切片的最大值
  2. CombineTextInputFormat.setMaxInputSplitSize(job,4194304);
  3. //设置使用其他的InputFormat
  4. job.setInputFormatClass(CombineTextInputFormat.class);

$06[Hadoop-MapResource框架原理(上)] - 图6

第二章.MapReducer工作流程

](https://imgtu.com/i/WEKWRO)

$06[Hadoop-MapResource框架原理(上)] - 图7

$06[Hadoop-MapResource框架原理(上)] - 图8

  1. # 上面的流程是整个MapReduce最全工作流程,但是shuffle过程只是从第7步开始到第16步结束,具体Shuffle过程详解如下:
  2. - （1）MapTask收集我们的map()方法输出的kv对，放到内存缓冲区中
  3. - （2）从内存缓冲区不断溢出本地磁盘文件，可能会溢出多个文件
  4. - （3）多个溢出文件会被合并成大的溢出文件
  5. - （4）在溢出过程及合并的过程中，都要调用Partitioner进行分区和针对key进行排序
  6. - （5）ReduceTask根据自己的分区号，去各个MapTask机器上取相应的结果分区数据
  7. - （6）ReduceTask会取到同一个分区的来自不同MapTask的结果文件，ReduceTask会将这些文件再进行合并（归并排序）
  8. - （7）合并成大文件后，Shuffle的过程也就结束了，后面进入ReduceTask的逻辑运算过程（从文件中取出一个一个的键值对Group，调用用户自定义的reduce()方法）
  1. # 注意
  2. 1. shuffle中的缓冲区大小可以通过参数调整,参数: io.sort.mb默认100M
  3. 2. shuffle缓冲区的大小会影响到MapReduce程序的执行效率,原则上说,缓冲区越大,磁盘io的次数越少,执行速度就越快

第三章.Shuffle 机制

Map方法之后,Reduce方法之前的数据处理过程称之为Shuffle

1.Shuffle机制

$06[Hadoop-MapResource框架原理(上)] - 图9

2.Partition分区

  1. # 1.问题引出
  2.   要求将统计结果按照输出条件输出到不同文件中(分区)
  3. # 2.默认Partitioner分区 
  4. 1.如果有多个ReduceTask时的默认分区：
  6. public class HashPartitioner<K, V> extends Partitioner<K, V> {
  8.   /*
  9.     作用 ：返回分区号
  10.     K : map输出的key
  11.     V : map输出的value
  12.     numReduceTasks : ReduceTask的数量
  13.   */
  14.   public int getPartition(K key, V value,
  15.                           int numReduceTasks) {
  17.     /*
  18.         分区号 = key.hashCode()  % numReduceTasks
  20.         key.hashCode() & Integer.MAX_VALUE ： 作用是用来保证结果为正数
  22.     */
  23.     return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
  24.   }
  28. 2.如果只有1个ReduceTask时的默认分区：
  29.         partitioner = new org.apache.hadoop.mapreduce.Partitioner<K,V>() {
  30.           @Override
  31.           public int getPartition(K key, V value, int numPartitions) {
  32.             return partitions - 1; //0
  33.           }
  34.         };
  1. # 3.自定义Parttitioner步骤
  2. - 自定义类继承Parttitioner,重写getPartition方法
  3. - 在job驱动中,设置自定义parttitioner
  4. job.setParttitionerClass(CustomPartitioner.class)
  5. - 自定义Partition后,要根据自定义Partitioner的逻辑设置相应数量的ReduceTask
  6. job.setNumReduceTasks(5)

$06[Hadoop-MapResource框架原理(上)] - 图10

3.Partition分区案例实操

需求:将统计结果按照手机归属地不同输出到不同文件中

需求分析

$06[Hadoop-MapResource框架原理(上)] - 图11

在之前案例的基础上,新增加一个分区类

  1. package com.atguigu.mr.partition;
  3. import org.apache.hadoop.io.Text;
  4. import org.apache.hadoop.mapreduce.Partitioner;
  5. /*
  6.       自定义分区类
  7.       1.自定义类并继承Partitioner
  8.            Partitioner<Key,Value>
  9.            Key:mapper写出的key的类型
  10.            value:mapper写出的value类型
  11.       2.重写getpartition方法
  12.  */
  13. public class MyPartitioner extends Partitioner <Text,FlowBean>{
  14.     /*
  15.     返回分区号
  16.      */
  17.     @Override
  18.     public int getPartition(Text text, FlowBean flowBean, int numPartitions) {
  19.         String phoneNumber = text.toString();
  20.         if(phoneNumber.startsWith("136")){
  21.             return 0;
  22.         }else if(phoneNumber.startsWith("137")){
  23.             return 1;
  24.         }else if(phoneNumber.startsWith("138")){
  25.             return 2;
  26.         }else if(phoneNumber.startsWith("139")){
  27.             return 3;
  28.         }else{
  29.             return 4;
  30.         }
  32.     }
  33. }
  1. package com.atguigu.mr.partition;
  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.         //设置ReduceTask的数量
  15.         job.setNumReduceTasks(5);
  16.         //设置自定义分区类
  17.         job.setPartitionerClass(MyPartitioner.class);
  18.         //给Job对象的属性赋值
  19.         job.setJarByClass(FlowDriver.class);
  20.         job.setMapperClass(FlowMapper.class);
  21.         job.setReducerClass(FlowReducer.class);
  22.         //2.3设置Mapper输出的k,v的类型
  23.         job.setMapOutputKeyClass(Text.class);
  24.         job.setMapOutputValueClass(FlowBean.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\\input4"));
  30.         //注意：输出路径不能存在！！！！！！！！！！！！
  31.         FileOutputFormat.setOutputPath(job,new Path("C:\\io\\output4"));
  33.         //3.提交Job对象（执行Job）
  34.         /*
  35.             参数 ：是否打印进度
  36.             返回值 ：如果是true则表示job执行成功
  37.          */
  38.         job.waitForCompletion(true);
  39.     }
  40. }

$06[Hadoop-MapResource框架原理(上)] - 图12