Hadoop学习：深入解析MapReduce的大数据魔力（三）

3.5 MapReduce 内核源码解析

3.5.1 MapTask 工作机制

1Read阶段MapTask通过InputFormat获得的RecordReader从输入InputSplit中解析出一个个key/value。

2Map阶段该节点主要是将解析出的key/value交给用户编写map()函数处理并产生一系列新的key/value。

3Collect 收集阶段在用户编写 map()函数中当数据处理完成后一般会调用OutputCollector.collect()输出结果。在该函数内部它会将生成的 key/value 分区调用Partitioner并写入一个环形内存缓冲区中。

4Spill 阶段即“溢写”当环形缓冲区满后MapReduce会将数据写到本地磁盘上生成一个临时文件。需要注意的是将数据写入本地磁盘之前先要对数据进行一次本地排序并在必要时对数据进行合并、压缩等操作。

溢写阶段详情
步骤1利用快速排序算法对缓存区内的数据进行排序排序方式是先按照分区编号Partition 进行排序然后按照key进行排序。这样经过排序后数据以分区为单位聚集在一起且同一分区内所有数据按照key有序。

步骤2按照分区编号由小到大依次将每个分区中的数据写入任务工作目录下的临时文件output/spillN.outN 表示当前溢写次数中。如果用户设置了 Combiner则写入文件之前对每个分区中的数据进行一次聚集操作。

步骤3将分区数据的元信息写到内存索引数据结构SpillRecord中其中每个分区的元信息包括在临时文件中的偏移量、压缩前数据大小和压缩后数据大小。如果当前内存索引大小超过1MB则将内存索引写到文件output/spillN.out.index中。

5Merge 阶段当所有数据处理完成后MapTask 对所有临时文件进行一次合并以确保最终只会生成一个数据文件。 当所有数据处理完后MapTask 会将所有临时文件合并成一个大文件并保存到文件output/file.out 中同时生成相应的索引文件output/file.out.index。 在进行文件合并过程中MapTask以分区为单位进行合并。对于某个分区它将采用多轮递归合并的方式。每轮合并mapreduce.task.io.sort.factor默认 10个文件并将产生的文件重新加入待合并列表中对文件排序后重复以上过程直到最终得到一个大文件。
让每个 MapTask 最终只生成一个数据文件可避免同时打开大量文件和同时读取大量小文件产生的随机读取带来的开销。

3.5.2 ReduceTask 工作机制

1Copy 阶段ReduceTask 从各个 MapTask 上远程拷贝一片数据并针对某一片数据如果其大小超过一定阈值则写到磁盘上否则直接放到内存中。

2Sort 阶段在远程拷贝数据的同时ReduceTask启动了两个后台线程对内存和磁盘上的文件进行合并以防止内存使用过多或磁盘上文件过多。按照MapReduce语义用户编写reduce()函数输入数据是按key进行聚集的一组数据。为了将key相同的数据聚在一
起Hadoop采用了基于排序的策略。由于各个MapTask已经实现对自己的处理结果进行了局部排序因此ReduceTask只需对所有数据进行一次归并排序即可。

3Reduce 阶段reduce()函数将计算结果写到HDFS上。

3.5.3 ReduceTask 并行度决定机制

回顾MapTask并行度由切片个数决定切片个数由输入文件和切片规则决定。

思考ReduceTask并行度由谁决定

1设置ReduceTask并行度个数
ReduceTask 的并行度同样影响整个 Job 的执行并发度和执行效率但与MapTask的并发数由切片数决定不同ReduceTask数量的决定是可以直接手动设置

// 默认值是1手动设置为4
job.setNumReduceTasks(4);

2实验测试ReduceTask多少合适
1实验环境1个Master节点16个Slave节点CPU:8GHZ内存: 2G
2实验结论

3注意事项
1ReduceTask=0表示没有Reduce阶段输出文件个数和Map个数一致。
2ReduceTask默认值就是1所以输出文件个数为一个。
3如果数据分布不均匀就有可能在Reduce阶段产生数据倾斜
4ReduceTask数量并不是任意设置还要考虑业务逻辑需求有些情况下需要计算全局汇总结果就只能有1个ReduceTask。
5具体多少个ReduceTask需要根据集群性能而定。
6如果分区数不是1但是ReduceTask为1是否执行分区过程。答案是不执行分区过程。因为在MapTask的源码中执行分区的前提是先判断ReduceNum个数是否大于1。不大于1肯定不执行。

3.6 数据清洗ETL

“ETL是英文 Extract-Transform-Load 的缩写用来描述将数据从来源端经过抽取
Extract、转换Transform、加载Load至目的端的过程。ETL一词较常用在数据仓库但其对象并不限于数据仓库

在运行核心业务MapReduce 程序之前往往要先对数据进行清洗清理掉不符合用户要求的数据。==清理的过程往往只需要运行Mapper程序不需要运行Reduce程序。 ==

1需求

去除日志中字段个数小于等于11的日志。
1输入数据

2期望输出数据
每行字段长度都大于11。

2需求分析

需要在Map阶段对输入的数据根据规则进行过滤清洗。

3实现代码

1编写WebLogMapper类

package com.atguigu.mapreduce.weblog; 
import java.io.IOException; 
import org.apache.hadoop.io.LongWritable; 
import org.apache.hadoop.io.NullWritable; 
import org.apache.hadoop.io.Text; 
import org.apache.hadoop.mapreduce.Mapper; 
 
public class WebLogMapper extends Mapper<LongWritable, Text, Text, 
NullWritable>{ 
  
 @Override 
 protected void map(LongWritable key, Text value, Context context) 
throws IOException, InterruptedException { 
   
  // 1 获取1行数据 
  String line = value.toString(); 
   
  // 2 解析日志 
  boolean result = parseLog(line,context); 
   
  // 3 日志不合法退出 
  if (!result) { 
   return; 
  } 
   
  // 4 日志合法就直接写出 
  context.write(value, NullWritable.get()); 
 } 
 
 // 2 封装解析日志的方法 
 private boolean parseLog(String line, Context context) { 
 
  // 1 截取 
  String[] fields = line.split(" "); 
   
  // 2 日志长度大于11的为合法 
  if (fields.length > 11) { 
   return true; 
  }else { 
   return false; 
  } 
 } 
} 

2编写WebLogDriver类

package com.atguigu.mapreduce.weblog; 
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; 
public class WebLogDriver { 
public static void main(String[] args) throws Exception { 
// 输入输出路径需要根据自己电脑上实际的输入输出路径设置 
args = new String[] { "D:/input/inputlog", "D:/output1" }; 
// 1 获取job信息 
Configuration conf = new Configuration(); 
Job job = Job.getInstance(conf); 
// 2 加载jar包 
job.setJarByClass(LogDriver.class); 
// 3 关联map 
job.setMapperClass(WebLogMapper.class); 
// 4 设置最终输出类型 
job.setOutputKeyClass(Text.class); 
job.setOutputValueClass(NullWritable.class); 
// 设置reducetask个数为0 
job.setNumReduceTasks(0); 
// 5 设置输入和输出路径 
FileInputFormat.setInputPaths(job, new Path(args[0])); 
FileOutputFormat.setOutputPath(job, new Path(args[1])); 
// 6 提交 
boolean b = job.waitForCompletion(true); 
System.exit(b ? 0 : 1); 
} 
} 

3.7 MapReduce 开发总结

1输入数据接口InputFormat

1默认使用的实现类是TextInputFormat
2TextInputFormat 的功能逻辑是一次读一行文本然后将该行的起始偏移量作为key行内容作为value返回。
3CombineTextInputFormat 可以把多个小文件合并成一个切片处理提高处理效率。

2逻辑处理接口Mapper

用户根据业务需求实现其中三个方法map() setup() cleanup ()

3Partitioner 分区

1有默认实现 HashPartitioner逻辑是根据key的哈希值和numReduces来返回一个分区号key.hashCode()&Integer.MAXVALUE % numReduces
2如果业务上有特别的需求可以自定义分区。

4Comparable 排序

1当我们用自定义的对象作为key来输出时就必须要实现WritableComparable 接口重写其中的compareTo()方法。

2部分排序对最终输出的每一个文件进行内部排序。

3全排序对所有数据进行排序通常只有一个Reduce。

4二次排序排序的条件有两个。

5Combiner 合并

Combiner 合并可以提高程序执行效率减少IO传输。但是使用时必须不能影响原有的
业务处理结果。

6逻辑处理接口Reducer

用户根据业务需求实现其中三个方法reduce() setup() cleanup ()

7输出数据接口OutputFormat

1默认实现类是TextOutputFormat功能逻辑是将每一个KV对向目标文本文件
输出一行。
2用户还可以自定义OutputFormat。