Hadoop3.x 之 MapReduce 框架原理(月薪过万第九章中)

朱雀 2022-09-02 01:14 166阅读 0赞

### Hadoop3.x 之 MapReduce 框架原理 ###

*  一、MapReduce 框架原理
 *   *  1、InputFormat 数据输入
     *   *  1) 问题引出
         *  2) MapTask 并行度决定机制
         *  3) 数据切片与MapTask并行度决定机制
         *  4) Job提交流程源码解析
         *  5) FileInputFormat切片源码解析
         *  6) FileInputFormat 切片机制
         *  7) FileInputFormat切片大小的参数配置
         *  8) TextInputFormat
         *  9) CombineTextInputFormat 切片机制
 *  二、MapReduce 工作流程
 *  三、Shuffle 机制
 *  四、Partition 分区
 *   *  1）问题引出
     *  2）默认Partitioner分区
     *  3）自定义Partitioner步骤
     *  4）Partition分区
     *  5）Partition 分区案例实操
 *  五、WritableComparable 排序
 *   *  1）排序概述
     *  2）排序分类
     *   *  ① 自定义排序 WritableComparable
         *  ② WritableComparable 排序案例实操（全排序）
         *   *  1、整体思路流程 实现步骤
             *  2、按照流程案例实现
         *  ③ WritableComparable 排序案例实操（区内排序）
         *   *  1、需求
             *  2、案例实操
         *  ④ Combiner 合并案例实操
         *   *  1、需求
             *  2、需求分析
             *  3、案例实操-方案一
             *  4、案例实操-方案二
 *  六、OutputFormat 接口实现类
 *   *  1）outputFormat接口实现类
     *  2）自定义 OutputFormat 案例实操
     *   *  1、需求
         *  2、需求分析
         *  3、案例实操

# 一、MapReduce 框架原理 #

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

*  InputFormat 可以对输入、进行一个控制、读出来之后、交给Mapper
 *  outputFormat 想要写到呐(不想输出到文件、可以到MySQL，ES) (就回去mapper拉取)

## 1、InputFormat 数据输入 ##

切片与 MapTask 并行度决定机制

### 1) 问题引出 ###

> MapTask 的并行度决定 Map 阶段的任务处理并发度，进而影响到整个 Job 的处理速度。

> 思考：1G 的数据，启动 8 个 MapTask，可以提高集群的并发处理能力。那么 1K 的数据，也启动 8 个 MapTask，会提高集群性能吗？MapTask 并行任务是否越多越好呢？哪些因素影响了 MapTask 并行度？

### 2) MapTask 并行度决定机制 ###

> 数据块：Block 是 HDFS 物理上把数据分成一块一块。数据块是 HDFS 存储数据单位。

> 数据切片：数据切片只是在逻辑上对输入进行分片，并不会在磁盘上将其切分成片进行存储。数据切片是 MapReduce 程序计算输入数据的单位，一个切片会对应启动一个 MapTask。

### 3) 数据切片与MapTask并行度决定机制 ###

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

### 4) Job提交流程源码解析 ###

Job 提交流程源码详解

waitForCompletion()
    submit();
    // 1 建立连接
    connect();
    // 1）创建提交 Job 的代理
    new Cluster(getConfiguration());
    // （1）判断是本地运行环境还是 yarn 集群运行环境
    initialize(jobTrackAddr, conf);
    // 2 提交 job
    submitter.submitJobInternal(Job.this, cluster)
    // 1）创建给集群提交数据的 Stag 路径
    Path jobStagingArea = JobSubmissionFiles.getStagingDir(cluster, conf);
    // 2）获取 jobid ，并创建 Job 路径
    JobID jobId = submitClient.getNewJobID();
    // 3）拷贝 jar 包到集群
    copyAndConfigureFiles(job, submitJobDir);
    rUploader.uploadFiles(job, jobSubmitDir);
    // 4）计算切片，生成切片规划文件
    writeSplits(job, submitJobDir);
    maps = writeNewSplits(job, jobSubmitDir);
    input.getSplits(job);
    // 5）向 Stag 路径写 XML 配置文件
    writeConf(conf, submitJobFile);
    conf.writeXml(out);
    // 6）提交 Job,返回提交状态
    status = submitClient.submitJob(jobId, submitJobDir.toString(),
    job.getCredentials());

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

### 5) FileInputFormat切片源码解析 ###

（1）程序先找到你数据存储的目录。
    （2）开始遍历处理（规划切片）目录下的每一个文件
    （3）遍历第一个文件ss.txt
         a）获取文件大小fs.sizeOf(ss.txt)
         b）计算切片大小
            computeSplitSize(Math.max(minSize,Math.min(maxSize,blocksize)))=blocksize=128M
         c）默认情况下，切片大小=blocksize
         d）开始切，形成第1个切片：ss.txt—0:128M 第2个切片ss.txt—128:256M 第3个切片ss.txt—256M:300M
     （每次切片时，都要判断切完剩下的部分是否大于块的1.1倍，不大于1.1倍就划分一块切片）
         e）将切片信息写到一个切片规划文件中
         f）整个切片的核心过程在getSplit()方法中完成
         g）InputSplit只记录了切片的元数据信息，比如起始位置、长度以及所在的节点列表等。
    （4）提交切片规划文件到YARN上，YARN上的MrAppMaster就可以根据切片规划文件计算开启MapTask个数。

### 6) FileInputFormat 切片机制 ###

FileInputFormat切片机制

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

### 7) FileInputFormat切片大小的参数配置 ###

> （1）源码中计算切片大小的公式  
> Math.max(minSize, Math.min(maxSize, blockSize));  
> mapreduce.input.fileinputformat.split.minsize=1 默认值为1  
> mapreduce.input.fileinputformat.split.maxsize= Long.MAXValue 默认值Long.MAXValue  
> 因此，默认情况下，切片大小=blocksize。

> （2）切片大小设置  
> maxsize（切片最大值）：参数如果调得比blockSize小，则会让切片变小，而且就等于配置的这个参数的值。  
> minsize（切片最小值）：参数调的比blockSize大，则可以让切片变得比blockSize还大。

> （3）获取切片信息API  
> // 获取切片的文件名称  
> String name = inputSplit.getPath().getName();  
> // 根据文件类型获取切片信息  
> FileSplit inputSplit = (FileSplit) context.getInputSplit();

### 8) TextInputFormat ###

FileInputFormat 实现类

1）FileInputFormat 实现类
    思考：在运行 MapReduce 程序时，输入的文件格式包括：基于行的日志文件、二进制格
    式文件、数据库表等。那么，针对不同的数据类型，MapReduce 是如何读取这些数据的呢？
        FileInputFormat 常见的接口实现类包括：
           TextInputFormat、KeyValueTextInputFormat、NLineInputFormat、
           CombineTextInputFormat 和自定义 InputFormat 等。
    
    2）TextInputFormat
    TextInputFormat 是默认的 FileInputFormat 实现类。按行读取每条记录。
    键是存储该行在整个文件中的起始字节偏移量， LongWritable 类型。值是这行的
    内容，不包括任何行终止符（换行符和回车符），Text 类型。
         以下是一个示例，比如，一个分片包含了如下 4 条文本记录。
         Rich learning form
         Intelligent learning engine
         Learning more convenient
         From the real demand for more close to the enterprise
         每条记录表示为以下键/值对：
          (0,Rich learning form)
          (20,Intelligent learning engine)
          (49,Learning more convenient)
          (74,From the real demand for more close to the enterprise)

### 9) CombineTextInputFormat 切片机制 ###

> 框架默认的 TextInputFormat 切片机制是对任务按文件规划切片，不管文件多小，都会是一个单独的切片，都会交给一个 MapTask，这样如果有大量小文件，就会产生大量的MapTask，处理效率极其低下。

> 1）应用场景：  
> CombineTextInputFormat 用于小文件过多的场景，它可以将多个小文件从逻辑上规划到一个切片中，这样，多个小文件就可以交给一个 MapTask 处理。

> 2）虚拟存储切片最大值设置  
> CombineTextInputFormat.setMaxInputSplitSize(job, 4194304);// 4m  
> 注意：虚拟存储切片最大值设置最好根据实际的小文件大小情况来设置具体的值。

> 3）切片机制  
> 生成切片过程包括：虚拟存储过程和切片过程二部分。

# 二、MapReduce 工作流程 #

MapReduce详细工作流程（一）  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyMDgyNzAx_size_16_color_FFFFFF_t_70 4]

MapReduce详细工作流程（二）

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

**上面的流程是整个 MapReduce 最全工作流程，但是 Shuffle 过程只是从第 7 步开始到第16 步结束，具体 Shuffle 过程详解，如下：**

*  （1）MapTask 收集我们的 map()方法输出的 kv 对，放到内存缓冲区中
 *  （2）从内存缓冲区不断溢出本地磁盘文件，可能会溢出多个文件
 *  （3）多个溢出文件会被合并成大的溢出文件
 *  （4）在溢出过程及合并的过程中，都要调用 Partitioner 进行分区和针对 key 进行排序
 *  （5）ReduceTask 根据自己的分区号，去各个 MapTask 机器上取相应的结果分区数据
 *  （6）ReduceTask 会抓取到同一个分区的来自不同 MapTask 的结果文件，ReduceTask 会将这些文件再进行合并（归并排序）
 *  （7）合并成大文件后，Shuffle 的过程也就结束了，后面进入 ReduceTask 的逻辑运算过程（从文件中取出一个一个的键值对 Group，调用用户自定义的 reduce()方法）

> 注意：  
> （1）Shuffle 中的缓冲区大小会影响到 MapReduce 程序的执行效率，原则上说，缓冲区  
> 越大，磁盘 io 的次数越少，执行速度就越快。  
> （2）缓冲区的大小可以通过参数调整，参数：mapreduce.task.io.sort.mb 默认 100M。

# 三、Shuffle 机制 #

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

# 四、Partition 分区 #

## 1）问题引出 ##

> 要求将统计结果按照条件输出到不同文件中（分区）。比如：将统计结果按照手机归属地不同省份输出到不同文件中（分区）

## 2）默认Partitioner分区 ##

public class HashPartitioner<K, V> extends Partitioner<K, V> { 
 public int getPartition(K key, V value, int numReduceTasks) { 
 return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
 }
 }

> 默认分区是根据key的hashCode对ReduceTasks个数取模得到的。用户没法控制哪个key存储到哪个分区

## 3）自定义Partitioner步骤 ##

把不同的数据分到自己想分区域

（1）自定义类继承Partitioner，重写getPartition()方法
 
 public class CustomPartitioner extends Partitioner<Text, FlowBean> { 
 @Override
 public int getPartition(Text key, FlowBean value, int numPartitions) { 
 // 控制分区代码逻辑
 return partition;
 }
 } 
 
 （2）在Job驱动中，设置自定义Partitioner
 job.setPartitionerClass(CustomPartitioner.class);
 
 （3）自定义Partition后，要根据自定义Partitioner的逻辑设置相应数量的ReduceTask
 job.setNumReduceTasks(5); #默认是1

## 4）Partition分区 ##

* 4.1 分区总结 
 （1）如果ReduceTask的数量> getPartition的结果数，则会多产生几个空的输出文件part-r-000xx； 
 （2）如果1<ReduceTask的数量<getPartition的结果数，则有一部分分区数据无处安放，会Exception； 
 （3）如 果ReduceTask的数量=1，则不管MapTask端输出多少个分区文件，最终结果都交给这一个 
 ReduceTask，最终也就只会产生一个结果文件 part-r-00000；
 * 4.2、案例分析 
 例如：假设自定义分区数为5，则

1.  job.setNumReduceTasks(1); 会正常运行，只不过会产生一个输出文件
2.  job.setNumReduceTasks(2); 会报错
3.  job.setNumReduceTasks(6); 大于5，程序会正常运行，会产生空文件

## 5）Partition 分区案例实操 ##

![在这里插入图片描述][8ee3160669ad3a7a85d1828ea8bfbbf6.png]

![在这里插入图片描述][27dd0a877db3d1de46475ad261a43745.png]

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

public class ProvincePartitioner extends Partitioner<Text, FlowBean> { 
 @Override
 public int getPartition(Text text, FlowBean flowBean, int numPartitions) { 
 String phone = text.toString();
 String prePhone = phone.substring(0, 3);
 int partition;
 
 // 自定义开启 5个分区
 if ("136".equals(prePhone)) { 
 partition = 0;
 } else if ("137".equals(prePhone)) { 
 partition = 1;
 } else if ("138".equals(prePhone)) { 
 partition = 2;
 } else if ("139".equals(prePhone)) { 
 partition = 3;
 } else { 
 partition = 4;
 }
 
 return partition;
 }
 }

在驱动函数中增加自定义数据分区设置和 ReduceTask 设置

public class FlowDriver { 
 public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { 
 //1、获取job
 Configuration config = new Configuration();
 Job job = Job.getInstance(config);
 
 //2、设置jar
 job.setJarByClass(FlowDriver.class);
 
 //3、关联mapper 和 reducer
 job.setMapperClass(FlowMapper.class);
 job.setReducerClass(FlowReducer.class);
 
 //4、设置mapper 输出的key和value 类型
 job.setMapOutputKeyClass(Text.class);
 job.setMapOutputValueClass(FlowBean.class);
 
 //5、设置最终数据输出的key和value 类型
 job.setOutputKeyClass(Text.class);
 job.setOutputValueClass(FlowBean.class);
 
 //###############################加上这两个配置
 job.setPartitionerClass(ProvincePartitioner.class); //跟自己写的建立连接
 //TODO 1、ReduceTasks 大于 我自定义的ProvincePartitioner 结果数、则会多产生几个空的输出文件part-r-000xx
 //TODO 2、如果1<ReduceTask的数量<getPartition的结果数，则有一部分分区数据无处安放，会Exception
 //TODO 3、如 果ReduceTask的数量=1，则不管MapTask端输出多少个分区文件，最终结果都交给这一个 ReduceTask，最终也就只会产生一个结果文件 part-r-00000；
 //TODO 4、分区号必须从零开始，逐一累加。
 /** * 例如：假设自定义分区数为5，则 * （1）job.setNumReduceTasks(1); 会正常运行，只不过会产生一个输出文件 * （ 2）job.setNumReduceTasks(2); 会报错 * （3）job.setNumReduceTasks(6); 大于5，程序会正常运行，会产生空文件 * * */
 job.setNumReduceTasks(5); //设置文本数5个
 
 //###############################
 //6、设置数据的输入和输出路径
 FileInputFormat.setInputPaths(job, new Path("D:\\input\\inputword"));
 FileOutputFormat.setOutputPath(job, new Path("D:\\hadoop\\output511111"));
 
 //7、提交job
 boolean result = job.waitForCompletion(true);
 System.exit(result ? 0 : 1);
 
 }
 }

# 五、WritableComparable 排序 #

## 1）排序概述 ##

> 排序是MapReduce框架中最重要的操作之一。

> 1.  MapTask和ReduceTask均会对数据按 照key进行排序。该操作属于  
>     Hadoop的默认行为。任何应用程序中的数据均会被排序，而不管逻辑上是  
>     否需要。

> 1.  默认排序是按照字典顺序排序，且实现该排序的方法是快速排序、

> 1.  对于MapTask，它会将处理的结果暂时放到环形缓冲区中，当环形缓冲区使  
>     用率达到一定阈值后，再对缓冲区中的数据进行一次快速排序，并将这些有序数 据溢写到磁盘上，而当数据处理完毕后，它会对磁盘上所有文件进行归并排序。

> 1.  对于ReduceTask，它从每个MapTask上远程拷贝相应的数据文件，如果文件大小超过一定阈值，则溢写磁盘上，否则存储在内存中。如果磁盘上文件数目达到一定阈值，则进行一次归并排序以生成一个更大文件；如果内存中文件大小或者数目超过一定阈值，则进行一次合并后将数据溢写到磁盘上。当所有数据拷贝完毕后，ReduceTask统一对内存和磁盘上的所有数据进行一次归并排序。

## 2）排序分类 ##

1.  部分排序  
    MapReduce根据输入记录的键对数据集排序。保证输出的每个文件内部有序。
2.  全排序  
    最终输出结果只有一个文件，且文件内部有序。实现方式是只设置一个ReduceTask。但该方法在  
    处理大型文件时效率极低，因为一台机器处理所有文件，完全丧失了MapReduce所提供的并行架构。
3.  辅助排序：（GroupingComparator分组） 在Reduce端对key进行分组。应用于：在接收的key为bean对象时，想让一个或几个字段相同（全部  
    字段比较不相同）的key进入到同一个reduce方法时，可以采用分组排序。
4.  二次排序  
    在自定义排序过程中，如果compareTo中的判断条件为两个即为二次排序。

### ① 自定义排序 WritableComparable ###

自定义排序 WritableComparable 原理分析  
bean 对象做为 key 传输，需要实现 WritableComparable 接口重写 compareTo 方法，就可以实现排序。

@Override
 public int compareTo(FlowBean bean) { 
 int result;
 // 按照总流量大小，倒序排列
 if (this.sumFlow > bean.getSumFlow()) { 
 result = -1;
 }else if (this.sumFlow < bean.getSumFlow()) { 
 result = 1;
 }else { 
 result = 0;
 }
 return result; }

### ② WritableComparable 排序案例实操（全排序） ###

#### 1、整体思路流程 实现步骤 ####

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

#### 2、按照流程案例实现 ####

*  FlowBean、实现WritableComparable接口 重写排序、包含二次排序

package org.example.wratablecomparable;
 
 import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.io.WritableComparable;
 
 import java.io.DataInput;
 import java.io.DataOutput;
 import java.io.IOException;
 
 /** * @ClassName FlowBean * @Author 小坏 * @Date 2021/7/16、16:54 * @Version 1.0 * * Hadoop 序列化 步骤 * 1、定义类实现 Writable 接口 * 2、重写序列化和反序列方法 * 3、重写空参构造 * 4、toString */
 
 //TODO 1
 public class FlowBean implements WritableComparable<FlowBean> { 
 
 private long upFlow; //上行流量
 private long downFlow; //下行流量
 private long sumFlow; //总流量
 
 
 //TODO 3
 public FlowBean() { 
 }
 
 // 序列化
 @Override
 public void write(DataOutput out) throws IOException { 
 out.writeLong(upFlow);
 out.writeLong(downFlow);
 out.writeLong(sumFlow);
 }
 
 //反序列化
 @Override
 public void readFields(DataInput in) throws IOException { 
 this.upFlow = in.readLong();
 this.downFlow = in.readLong();
 this.sumFlow = in.readLong();
 }
 
 
 public long getUpFlow() { 
 return upFlow;
 }
 
 public void setUpFlow(long upFlow) { 
 this.upFlow = upFlow;
 }
 
 public long getDownFlow() { 
 return downFlow;
 }
 
 public void setDownFlow(long downFlow) { 
 this.downFlow = downFlow;
 }
 
 public long getSumFlow() { 
 return sumFlow;
 }
 
 public void setSumFlow(long sumFlow) { 
 this.sumFlow = sumFlow;
 }
 
 //总流量
 public void setSumFlow() { 
 this.sumFlow = this.downFlow + this.upFlow;
 }
 
 //改造ToString的输出
 @Override
 public String toString() { 
 return upFlow + "\t" + downFlow + "\t" + sumFlow;
 }
 
 @Override
 public int compareTo(FlowBean o) { 
 //按照总流量比较,倒序排列
 if (this.sumFlow > o.sumFlow) { 
 return -1;
 } else if (this.sumFlow < o.sumFlow) { 
 return 1;
 } else { 
 //按照上行流量的正序排
 if (this.upFlow > o.upFlow) { 
 return -1;
 } else if (this.upFlow < o.upFlow) { 
 return 1;
 } else { 
 return 0;
 }
 }
 }
 
 /** * 输出格式： * * 1116 954 2070 * 上行流量 下行流量 总流量 */
 }

FlowMapper、实现获取一行、切割

package org.example.wratablecomparable;
 
 import org.apache.hadoop.io.LongWritable;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.mapreduce.Mapper;
 
 import java.io.IOException;
 
 /** * @ClassName FlowMapper * @Author 小坏 * @Date 2021/7/16、17:10 * @Version 1.0 * * 1、输入的 K,V * LongWritable, Text(一行) * * 1、输出的 K,V * Text(手机号), FlowBean(FlowBean对象) */
 public class FlowMapper extends Mapper<LongWritable, Text, FlowBean, Text> { 
 
 private FlowBean outV = new FlowBean();
 private Text outK = new Text();
 
 @Override
 protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { 
 //获取一行
 String line = value.toString();
 //切割
 String[] split = line.split("\t");
 
 outK.set(split[0]);
 outV.setUpFlow(Long.parseLong(split[1]));
 outV.setDownFlow(Long.parseLong(split[2]));
 outV.setSumFlow();
 
 //4 写出 outK outV
 context.write(outV,outK);
 }
 
 
 }

*  FlowReducer 收集

package org.example.wratablecomparable;
 
 
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.mapreduce.Reducer;
 
 import java.io.IOException;
 
 /** * @ClassName FlowReducer * @Author 小坏 * @Date 2021/7/16、17:58 * @Version 1.0 * * 1、Reducer的 k,v * 就对应这 map的输出 Text(电话号码), FlowBean(Bean对象) * * 2、最终输出的也是 Text, FlowBean */
 public class FlowReducer extends Reducer<FlowBean, Text, Text, FlowBean> { 
 
 private FlowBean outV = new FlowBean();
 
 @Override
 protected void reduce(FlowBean key, Iterable<Text> values, Context context) throws IOException, InterruptedException { 
 for (Text value : values) { 
 context.write(value, key);
 }
 }
 }

*  FlowDriver 关联Mapper 和 Reducer

package org.example.wratablecomparable;
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Job;
    import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    
    import java.io.IOException;
    
    /** * @ClassName FlowDriver * @Author 小坏 * @Date 2021/7/16、18:13 * @Version 1.0 */
    public class FlowDriver { 
        public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { 
            //1、获取job
            Configuration config = new Configuration();
            Job job = Job.getInstance(config);
    
            //2、设置jar
            job.setJarByClass(FlowDriver.class);
    
            //3、关联mapper 和 reducer
            job.setMapperClass(FlowMapper.class);
            job.setReducerClass(FlowReducer.class);
    
            //4、设置mapper 输出的key和value 类型
            job.setMapOutputKeyClass(FlowBean.class);
            job.setMapOutputValueClass(Text.class);
    
            //5、设置最终数据输出的key和value 类型
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(FlowBean.class);
            //6、设置数据的输入和输出路径- >输出数据 是上一个序列化的结果
            FileInputFormat.setInputPaths(job, new Path("D:\\hadoop\\out1put61061"));
            FileOutputFormat.setOutputPath(job, new Path("D:\\hadoop\\out1put61062"));
    
            //7、提交job
            boolean result = job.waitForCompletion(true);
    
            System.exit(result ? 0 : 1);
    
        }
    }

### ③ WritableComparable 排序案例实操（区内排序） ###

#### 1、需求 ####

1.  需求  
    要求每个省份手机号输出的文件中按照总流量内部排序。
2.  需求分析  
    基于前一个需求，增加自定义分区类，分区按照省份手机号设置。

*  分区内排序案例分析  
    ![在这里插入图片描述][2c24f6627a0c91846c40e63e01fcdad1.png]

#### 2、案例实操 ####

FlowBean 操作对象

package org.example.partitionerandwritableComparable;
 
 import org.apache.hadoop.io.WritableComparable;
 
 import java.io.DataInput;
 import java.io.DataOutput;
 import java.io.IOException;
 
 /** * @ClassName FlowBean * @Author 小坏 * @Date 2021/7/16、16:54 * @Version 1.0 * * Hadoop 序列化 步骤 * 1、定义类实现 Writable 接口 * 2、重写序列化和反序列方法 * 3、重写空参构造 * 4、toString */
 
 //TODO 1
 public class FlowBean implements WritableComparable<FlowBean> { 
 
 private long upFlow; //上行流量
 private long downFlow; //下行流量
 private long sumFlow; //总流量
 
 
 //TODO 3
 public FlowBean() { 
 }
 
 // 序列化
 @Override
 public void write(DataOutput out) throws IOException { 
 out.writeLong(upFlow);
 out.writeLong(downFlow);
 out.writeLong(sumFlow);
 }
 
 //反序列化
 @Override
 public void readFields(DataInput in) throws IOException { 
 this.upFlow = in.readLong();
 this.downFlow = in.readLong();
 this.sumFlow = in.readLong();
 }
 
 
 public long getUpFlow() { 
 return upFlow;
 }
 
 public void setUpFlow(long upFlow) { 
 this.upFlow = upFlow;
 }
 
 public long getDownFlow() { 
 return downFlow;
 }
 
 public void setDownFlow(long downFlow) { 
 this.downFlow = downFlow;
 }
 
 public long getSumFlow() { 
 return sumFlow;
 }
 
 public void setSumFlow(long sumFlow) { 
 this.sumFlow = sumFlow;
 }
 
 //总流量
 public void setSumFlow() { 
 this.sumFlow = this.downFlow + this.upFlow;
 }
 
 //改造ToString的输出
 @Override
 public String toString() { 
 return upFlow + "\t" + downFlow + "\t" + sumFlow;
 }
 
 @Override
 public int compareTo(FlowBean o) { 
 //按照总流量比较,倒序排列
 if (this.sumFlow > o.sumFlow) { 
 return -1;
 } else if (this.sumFlow < o.sumFlow) { 
 return 1;
 } else { 
 //按照上行流量的正序排
 if (this.upFlow > o.upFlow) { 
 return -1;
 } else if (this.upFlow < o.upFlow) { 
 return 1;
 } else { 
 return 0;
 }
 }
 }
 
 /** * 输出格式： * * 1116 954 2070 * 上行流量 下行流量 总流量 */
 }

FlowMapper 分析数据

package org.example.partitionerandwritableComparable;
 
 import org.apache.hadoop.io.LongWritable;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.mapreduce.Mapper;
 
 import java.io.IOException;
 
 /** * @ClassName FlowMapper * @Author 小坏 * @Date 2021/7/16、17:10 * @Version 1.0 * * 1、输入的 K,V * LongWritable, Text(一行) * * 1、输出的 K,V * Text(手机号), FlowBean(FlowBean对象) */
 public class FlowMapper extends Mapper<LongWritable, Text, FlowBean, Text> { 
 
 private FlowBean outV = new FlowBean();
 private Text outK = new Text();
 
 @Override
 protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { 
 //获取一行
 String line = value.toString();
 //切割
 String[] split = line.split("\t");
 
 outK.set(split[0]);
 outV.setUpFlow(Long.parseLong(split[1]));
 outV.setDownFlow(Long.parseLong(split[2]));
 outV.setSumFlow();
 
 //4 写出 outK outV
 context.write(outV,outK);
 }
 
 
 }

FlowReducer 收集数据

package org.example.partitionerandwritableComparable;
 
 
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.mapreduce.Reducer;
 
 import java.io.IOException;
 
 /** * @ClassName FlowReducer * @Author 小坏 * @Date 2021/7/16、17:58 * @Version 1.0 * * 1、Reducer的 k,v * 就对应这 map的输出 Text(电话号码), FlowBean(Bean对象) * * 2、最终输出的也是 Text, FlowBean */
 public class FlowReducer extends Reducer<FlowBean, Text, Text, FlowBean> { 
 
 
 @Override
 protected void reduce(FlowBean key, Iterable<Text> values, Context context) throws IOException, InterruptedException { 
 for (Text value : values) { 
 context.write(value, key);
 }
 }
 }

ProvincePartitioner2 设置分区

package org.example.partitionerandwritableComparable;
 
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.mapreduce.Partitioner;
 
 /** * 设置分区 * * @ClassName ProvincePartitioner2 * @Author 小坏 * @Date 2021/7/21、17:07 * @Version 1.0 */
 public class ProvincePartitioner2 extends Partitioner<FlowBean, Text> { 
 
 @Override
 public int getPartition(FlowBean flowBean, Text text, int numPartitions) { 
 String phone = text.toString();
 
 String prePhone = phone.substring(0, 3);
 
 int partition;
 
 if ("136".equals(prePhone)) { 
 partition = 0;
 } else if ("137".equals(prePhone)) { 
 partition = 1;
 } else if ("138".equals(prePhone)) { 
 partition = 2;
 } else if ("139".equals(prePhone)) { 
 partition = 3;
 } else { 
 partition = 4;
 }
 return partition;
 }
 }

FlowDriver 操作类

package org.example.partitionerandwritableComparable;
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Job;
    import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    
    import java.io.IOException;
    
    /** * WritableComparable 排序案例实操（区内排序） * 包含二次排序 * * 1）需求 * 要求每个省份手机号输出的文件中按照总流量内部排序。 * 2）需求分析 * 基于前一个需求，增加自定义分区类，分区按照省份手机号设置。 * @ClassName FlowDriver * @Author 小坏 * @Date 2021/7/16、18:13 * @Version 1.0 */
    public class FlowDriver { 
        public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { 
            //1、获取job
            Configuration config = new Configuration();
            Job job = Job.getInstance(config);
    
            //2、设置jar
            job.setJarByClass(FlowDriver.class);
    
            //3、关联mapper 和 reducer
            job.setMapperClass(FlowMapper.class);
            job.setReducerClass(FlowReducer.class);
    
            //4、设置mapper 输出的key和value 类型
            job.setMapOutputKeyClass(FlowBean.class);
            job.setMapOutputValueClass(Text.class);
    
            job.setPartitionerClass(ProvincePartitioner2.class);
            job.setNumReduceTasks(5);
    
            //5、设置最终数据输出的key和value 类型
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(FlowBean.class);
            //6、设置数据的输入和输出路径- >输出数据 是上一个序列化的结果
            FileInputFormat.setInputPaths(job, new Path("D:\\hadoop\\output4"));
            FileOutputFormat.setOutputPath(job, new Path("D:\\hadoop\\output6"));
    
            //7、提交job
            boolean result = job.waitForCompletion(true);
    
            System.exit(result ? 0 : 1);
    
        }
    }

### ④ Combiner 合并案例实操 ###

#### 1、需求 ####

统计过程中对每一个 MapTask 的输出进行局部汇总，以减小网络传输量即采用  
Combiner 功能。  
![在这里插入图片描述][307fc199cda3a15877421a4c41d24310.png]

#### 2、需求分析 ####

![在这里插入图片描述][1e32631e65525ee14fb972531e015e87.png]

#### 3、案例实操-方案一 ####

（1）增加一个 WordCountCombiner 类继承 Reducer

package com.atguigu.mapreduce.combiner;
 import org.apache.hadoop.io.IntWritable;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.mapreduce.Reducer;
 import java.io.IOException;
 public class WordCountCombiner extends Reducer<Text, IntWritable, Text, IntWritable> { 
 private IntWritable outV = new IntWritable();
 @Override
 protected void reduce(Text key, Iterable<IntWritable> values, Context 
 context) throws IOException, InterruptedException { 
 int sum = 0;
 for (IntWritable value : values) { 
 sum += value.get();
 }
 //封装 outKV
 outV.set(sum);
 //写出 outKV
 context.write(key,outV);
 } }

（2）在 WordcountDriver 驱动类中指定 Combiner

// 指定需要使用 combiner，以及用哪个类作为 combiner 的逻辑  
job.setCombinerClass(WordCountCombiner.class);

#### 4、案例实操-方案二 ####

设置 Combiner

job.setCombinerClass(WordCountReducer.class); 在map阶段就帮我们处理了一下  
![在这里插入图片描述][1e35a17581193c9114adb0c0db354cd3.png]

源码就不展示了、信息量有点大

# 六、OutputFormat 接口实现类 #

## 1）outputFormat接口实现类 ##

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

## 2）自定义 OutputFormat 案例实操 ##

### 1、需求 ###

过滤输入的 log 日志，包含 atguigu 的网站输出到 e:/atguigu.log，不包含 atguigu 的网站输出到 e:/other.log。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyMDgyNzAx_size_16_color_FFFFFF_t_70 8]

### 2、需求分析 ###

自定义OutputFormat案例分析

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

### 3、案例实操 ###

LogDriver

package org.example.outputformat;
    
    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.FileInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    
    import java.io.IOException;
    
    /** * OutputFormat案例分析 * * @ClassName LogDriver * @Author 小坏 * @Date 2021/7/23、11:54 * @Version 1.0 */
    public class LogDriver { 
        public static void main(String[] args) throws IOException,
                ClassNotFoundException, InterruptedException { 
            Configuration conf = new Configuration();
            Job job = Job.getInstance(conf);
            job.setJarByClass(LogDriver.class);
            job.setMapperClass(LogMapper.class);
            job.setReducerClass(LogReducer.class);
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(NullWritable.class);
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(NullWritable.class);
    
            //设置自定义的 outputformat
            job.setOutputFormatClass(LogOutPutFormat.class);
            FileInputFormat.setInputPaths(job, new Path("D:\\input\\inputoutputformat"));
    
            // 虽 然 我 们 自 定 义 了 outputformat ， 但 是 因 为 我 们 的 outputformat 继承自 fileoutputformat
            //而 fileoutputformat 要输出一个_SUCCESS 文件，所以在这还得指定一个输出目录
            FileOutputFormat.setOutputPath(job, new Path("D:\\hadoop\\output1111"));
    
            boolean b = job.waitForCompletion(true);
            System.exit(b ? 0 : 1);
        }
    }

LogMapper

package org.example.outputformat;
 
 import org.apache.hadoop.io.LongWritable;
 import org.apache.hadoop.io.NullWritable;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.mapreduce.Mapper;
 
 import java.io.IOException;
 
 /** * 自定义输出文件格式 * * @ClassName LogMapper * @Author 小坏 * @Date 2021/7/23、10:47 * @Version 1.0 * * LongWritable 输入进来第一个k v：Text * * 第二个 输出的K 就是它这一行的内容 v:里面没有、就用 NullWritable */
 public class LogMapper extends Mapper<LongWritable, Text, Text, NullWritable> { 
 
 @Override
 protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { 
 /** * 进来是这中数据格式 * http://www.baidu.com * http://www.google.com */
 
 //不做任何处理、来一行、写出一行
 context.write(value, NullWritable.get());
 
 
 }
 }

LogOutPutFormat

package org.example.outputformat;
 
 import org.apache.hadoop.io.NullWritable;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.mapreduce.RecordWriter;
 import org.apache.hadoop.mapreduce.TaskAttemptContext;
 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
 
 import java.io.IOException;
 
 /** * 自定义输出文件、 * @ClassName LogOutPutFormat * @Author 小坏 * @Date 2021/7/23、11:02 * @Version 1.0 */
 public class LogOutPutFormat extends FileOutputFormat<Text, NullWritable> { 
 @Override
 public RecordWriter<Text, NullWritable> getRecordWriter(TaskAttemptContext job) throws IOException, InterruptedException { 
 
 //返回值是RecordWriter<Text, NullWritable> 那就去创建一个
 LogRecordWriter lrw =new LogRecordWriter(job);
 return lrw;
 }
 }

LogRecordWriter

package org.example.outputformat;
 
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.io.IOUtils;
 import org.apache.hadoop.io.NullWritable;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.mapreduce.RecordWriter;
 import org.apache.hadoop.mapreduce.TaskAttemptContext;
 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
 
 import java.io.IOException;
 
 /** * @ClassName LogRecordWriter * @Author 小坏 * @Date 2021/7/23、11:06 * @Version 1.0 */
 public class LogRecordWriter extends RecordWriter<Text, NullWritable> { 
 
 private FSDataOutputStream atguiguOut;
 private FSDataOutputStream otherOut;
 
 public LogRecordWriter(TaskAttemptContext job) { 
 try { 
 //创建两个流、需要声明全局的才可以 让别人具体写
 FileSystem fs = FileSystem.get(job.getConfiguration());
 atguiguOut = fs.create(new Path("D:\\hadoop\\atguigu.log"));
 otherOut = fs.create(new Path("D:\\hadoop\\other.log"));
 } catch (IOException e) { 
 
 e.printStackTrace();
 }
 }
 
 /** * 写出 * * @param key * @param value * @throws IOException * @throws InterruptedException */
 @Override
 public void write(Text key, NullWritable value) throws IOException, InterruptedException { 
 String log = key.toString();
 if (log.contains("atguigu")) { 
 atguiguOut.writeBytes(log + "\n");
 } else { 
 otherOut.writeBytes(log + "\n");
 }
 }
 
 @Override
 public void close(TaskAttemptContext context) throws IOException, InterruptedException { 
 //关流
 IOUtils.closeStream(atguiguOut);
 IOUtils.closeStream(otherOut);
 }
 }

LogReducer

package org.example.outputformat;
 
 import org.apache.hadoop.io.LongWritable;
 import org.apache.hadoop.io.NullWritable;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.mapreduce.Reducer;
 
 import java.io.IOException;
 
 /** * @ClassName LogReducer * @Author 小坏 * @Date 2021/7/23、10:55 * @Version 1.0 */
 public class LogReducer extends Reducer<Text, NullWritable, Text, NullWritable> { 
 @Override
 protected void reduce(Text key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException { 
 /** * 进来是这中数据格式 * http://www.baidu.com * http://www.google.com * * //防止有相同数据、丢数据 、做循环 */
 
 for (NullWritable value : values) { 
 context.write(key, NullWritable.get());
 }
 }
 }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyMDgyNzAx_size_16_color_FFFFFF_t_70]: /images/20220829/73f7b1cdfcee4e5098f44969ea9d1987.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyMDgyNzAx_size_16_color_FFFFFF_t_70 1]: /images/20220829/683a279d2d82411e816febad22e3eb49.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyMDgyNzAx_size_16_color_FFFFFF_t_70 2]: /images/20220829/e51164a5c4924a9f9bf60aafe7b2ba67.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyMDgyNzAx_size_16_color_FFFFFF_t_70 3]: /images/20220829/4eda4c69c7954449aab8d1226f1052fc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyMDgyNzAx_size_16_color_FFFFFF_t_70 4]: /images/20220829/1d25a9cb229d4123aa6f27fddb505a12.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyMDgyNzAx_size_16_color_FFFFFF_t_70 5]: /images/20220829/c980a67f9cb64d2586e5c4138aa3e3d8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyMDgyNzAx_size_16_color_FFFFFF_t_70 6]: /images/20220829/6dbe9a8a17fe4eac9d7c87635237ac98.png
[8ee3160669ad3a7a85d1828ea8bfbbf6.png]: /images/20220829/3a90f42e6fab46fd99dc4a82120e903c.png
[27dd0a877db3d1de46475ad261a43745.png]: /images/20220829/c3ddf415cd0a49c59e47d6f0a5d8983a.png
[a4e9f9cd95baf153777c6037c188b890.png]: /images/20220829/5f49dbfe4adb434290df23099d72078e.png
[2c24f6627a0c91846c40e63e01fcdad1.png]: /images/20220829/a487fda4b6b34b96b81979f75a27f48f.png
[307fc199cda3a15877421a4c41d24310.png]: /images/20220829/da886676c15e471aa0b33adafe962fdc.png
[1e32631e65525ee14fb972531e015e87.png]: /images/20220829/e404fb95d541421d86902c211299fce4.png
[1e35a17581193c9114adb0c0db354cd3.png]: /images/20220829/59deae80f16a42aabe524d72640efb4b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyMDgyNzAx_size_16_color_FFFFFF_t_70 7]: /images/20220829/d76f5a2e996148d89a272d7934db2848.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyMDgyNzAx_size_16_color_FFFFFF_t_70 8]: /images/20220829/e85c8c53572244a9b00283db2827bcf5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyMDgyNzAx_size_16_color_FFFFFF_t_70 9]: /images/20220829/3ebbd2e802c54881a05ab451ceb5a23a.png