mapreduce

灰太狼 2022-10-18 12:52 217阅读 0赞

**job：**

*  setNumReduces，设置reduce task数目；
 *  waitForCompletion-》submit提交任务
    
     *  connect，链接本地或yarn集群，创建Cluster-》initialize-》创建protocol
     *  submitJobInternal-》
        
        1.  检查job和目录checkSpecs
        2.  submitClient.getNewJobID，生成全局唯一ID
        3.  writeSplits-》获取逻辑分区写入hdfs，返回**maptask个数**，存入conf；
        4.  writeConf-》生成job.xml，把文件写到hdfs
        5.  submitJob-》ApplicationClientProtocol.submitApplication-》通知resource manager，提交任务，发送协议

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hd2VucWlhbmc_size_16_color_FFFFFF_t_70][]

**任务调度：**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hd2VucWlhbmc_size_16_color_FFFFFF_t_70 1][]

**yarn child：**

taskFinal.run执行maptask/reducetask；

**mapper：**

1.  maptask.run-》如果存在reduce task，设置map、sort占比；
2.  initialize-》初始化outputFormat；
3.  runNewMapper-》
    
    1.  生成mapper，input format，record reader input，record writer output，input split；
    2.  环形缓冲区初始化，createSortingCollector-》MapOutputCollector-》MapOutputBuffer，默认100M；
    3.  output初始化，NewOutputCollector-》如果reduce task数目大于1，生成partitioner；
    4.  封装到mapContext（input，output，split）；
    5.  再封装到Mapper.Context，改context就是mapper中使用的context；
4.  input.initialize-》CreateRecordReader，创建LineRecordReader，访问数据源；
5.  mapper.run-》调用我们定义的mapper，Mapper循环读取一行数据，调用Map处理，提取想要的数据，写入到输出流；
    
    1.  setup，做预处理
    2.  context.nextKeyValue()-》LineRecordReader.nextKeyValue
    3.  map-》**context.write**\-》WrappedMapper.write-》**NewOutputCollector**\-》write-》获取**Partitioner(分区规则）**，对k,v数据打上分区标记
    4.  默认HashPartitioner，(key.hashCode() & Integer.MAX\_VALUE) % numReduceTasks；
    5.  MapOutputBuffer.Collect，写入环形缓冲区
6.  spillThread负责对环形缓冲区溢出处理-》buffer达到一定阈值，sortAndSpill()-》
    
    1.  对buffer中数据，进行排序；(**排序规则Sorter**）
    2.  遍历分区，查看是否有**Combiner**，有的话，进行局部聚合，否则略过；
    3.  写入OutputConverter-》combineCollector-》写入文件；
    4.  spill文件里面会有多个分区数据；
7.  close-》MapOutputBuffer.flush-》MergeParts，按照分区对spill file的合并；
8.  生成一个数据文件和一个索引文件；
9.  SpillRecord管理多个IndexRecord（一个分区信息），然后写入索引文件；

shuffle：

包含三大组件，partitioner，sorter，combiner（默认该combiner就是reduce最终聚合的逻辑）

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hd2VucWlhbmc_size_16_color_FFFFFF_t_70 2][]

**MapTask并行度：**

FileInputFormat-》getSplits-》

1.  获取，List<InputSplit>，逻辑切片集合，FileSplit为具体实现；
2.  "mapreduce.input.fileinputformat.split.minsize"如果没配置，则默认1；
3.  遍历job要处理的file，如果文件可以切分
4.  根据block size，上面的配置，最大分割值（若没配置"mapreduce.input.fileinputformat.split.maxsize"就是long最大值）；
5.  computeSplitSize，用这三个值计算切片大小，取三者中间值；**通过调整最小值，最大值，来决定分片大小；**
6.  循环切片，如果剩余文件大小/splitSize > 1.1，那么一直循环，存入list；

元数据：

private Path file;
      private long start;
      private long length;
      private String[] hosts;

**reducer:**

1.  reduce task个数大于等于partitions个数；
2.  
3.  reduce，拿到相同组的数据，**分组规则GroupComparator**；
4.  shuffleConsumerPlugin拉取数据，数据管理有两个类InMemoryMapOutput，OnDiskMapOutput；
5.  内存达到阈值，spill，写入磁盘；
6.  finalMerge-》Merger.merge，Merger.writeFile，合并成一个文件；
7.  reducer.run-》从文件不断读取数据，**reduce**，

**mapper与reducer的join：**

大小表的情况可以再mapper做；

否则只能再reducer做；

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hd2VucWlhbmc_size_16_color_FFFFFF_t_70]: /images/20221005/59bf88bb7f514bf28f047098b1c68079.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hd2VucWlhbmc_size_16_color_FFFFFF_t_70 1]: /images/20221005/1ae5659faa104cfe9e3016e3ebf7eb39.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hd2VucWlhbmc_size_16_color_FFFFFF_t_70 2]: /images/20221005/4f19b6f4c03d45fd92d6d3e29a3e4e03.png