19.MapReduce五大过程

迷南。 2022-06-16 06:25 131阅读 0赞

https://www.zybuluo.com/awsekfozc/note/219974

`MapReduce`

## ![mapreduce五大过程.png-47kB][mapreduce_.png-47kB] ##

## Input ##

>  *  将输入日志数据，数据库数据输入到map函数 （key-value形式）
>  *  一行记录对于一个键值对（key-value）

## Map ##

> 1）每个节点都需要执行代码（需要打成jar包）   
> 2）对输入的数据做逻辑处理。   
> 3）接受一个键值对（key-value），产生一组中间键值对。map函数产生的中间键值对里键相同的值传递给一个reduce函数。   
> 4）map任务结束之前，将处理的结果放置再内存中（默认大小为100M），称之为“环形缓冲区”。当内存占到一定比率时（默认80%），把内存中的数据溢写到磁盘（spill）

## Shuffle ##

### map shuffle ###

> 1）分区partitioner：当map在内存中的数据溢写到磁盘之前。根据reduce的不同。自动将数据分为到不同的区域。决定了map输出的数据被那个对应的reduce处理。   
> 2）排序sorter：对分区中的数据排序，此时数据还在内存之中，排序速度快。   
> 3）溢写spill:上述过程完成后，将数据写到spill到本地磁盘。   
> 4）合并merge:当map处理数据结束以后。合并各个分区的数据，排序合并后的数据。形成一个分区完成排序完成的文件。   
> 5）map端的reduce（combiner）（可选，不是所有mapreduce都能设置combiner）   
> 6）压缩：压缩Map输出的数据

#### 参数设置 ####

<property>
       <name>mapreduce.map.cpu.vcores</name>
       <value>1</value>
       </property>
       
       <property>
       <name>mapreduce.reduce.cpu.vcores</name>
       <value>1</value>
       </property>
       
       <property>
       <name>mapreduce.task.io.sort.factor</name>
       <value>10</value>
       </property>
       
       <property>
       <name>mapreduce.task.io.sort.mb</name>
       <value>100</value>
       </property>
       
       <property>
       <name>mapreduce.map.sort.spill.percent</name>
       <value>0.80</value>
       </property>

#### 分区设置（partitioner） ####

/**
       * map shuffle 分区设置（partitioner）
       */
       package com.zc.hadoop.shuffle;
       import org.apache.hadoop.io.Text;
       import org.apache.hadoop.mapreduce.Partitioner;
       
       public class TestPartitioner extends Partitioner<Text, Text>{ 
       public int getPartition(Text arg0, Text arg1, int arg2) { 
       // TODO Auto-generated method stub
       return 0;
       }
       }
       
       
       /**
       * mapreduce中调用
       */
       job.setPartitionerClass(TestPartitioner.class);

#### 排序，分组设置（srot,group） ####

/**
       * map shuffle 排序，分组设置（srot,group）
       */
       package com.zc.hadoop.shuffle;
       
       import org.apache.hadoop.io.RawComparator;
       
       import com.zc.hadoop.mapreduce.WordCountMr;
       import com.zc.hadoop.mapreduce.WordCountMr.WCMapper;
       
       public class TestRawComparator implements RawComparator<WordCountMr.WCMapper> { 
       
       public int compare(WCMapper arg0, WCMapper arg1) { 
       // TODO Auto-generated method stub
       return 0;
       }
       
       public int compare(byte[] arg0, int arg1, int arg2, byte[] arg3, int arg4,
       int arg5) { 
       // TODO Auto-generated method stub
       return 0;
       }
       
       }
       
       /**
       * mapreduce中调用
       */
       job.setCombinerClass(TestRawComparator.class);
       job.setGroupingComparatorClass(TestRawComparator.class);

#### combiner设置 ####

/**
       * combiner
       */
       package com.zc.hadoop.shuffle;
       import java.io.IOException;
       import java.util.Iterator;
       import org.apache.hadoop.io.IntWritable;
       import org.apache.hadoop.io.Text;
       import org.apache.hadoop.mapred.OutputCollector;
       import org.apache.hadoop.mapred.Reporter;
       import org.apache.hadoop.mapreduce.Reducer;
       
       public class TestCombiner extends Reducer<Text, IntWritable, IntWritable, Text>{ 
       public void reduce(Text key, Iterator<IntWritable> values,
       OutputCollector<IntWritable, Text> output, Reporter reporter)
       throws IOException { 
       // TODO Auto-generated method stub
       
       }
       }
       
       /**
       * mapreduce中调用
       */
       job.setCombinerClass(WordCountCombiner.class);

##### combiner结果 #####

![c2.png-35.9kB][]

#### 压缩设置（comperss） ####

/**
       * 设置配置（代码设置方式）
       * param1:
       * param2:
       */
       /**
       * param1:启用压缩
       * param2:true 启用
       */
       configuration.set("mapreduce.map.output.compress", "true");
       
       /**
       * param1:设置压缩类型
       * param2:类路径
       */
       configuration.set( "mapreduce.map.output.compress.codec","org.apache.hadoop.io.compress.SnappyCodec");

<property>
       <name>mapreduce.map.output.compress</name>
       <value>true</value>
       </property>
       
       
       <property>
       <name>mapreduce.map.output.compress.codec</name>
       <value>org.apache.hadoop.io.compress.SnappyCodec</value>
       </property>

### reduce shuffle ###

> 1）读取数据：已完成的map task存储在本地磁盘的数据（map shuffle过程一将之，分组，合并，排序）。为各个reduce task去拉取每个reduce要处理的数据。将相同key的数据合并在一起，放置再内存中。

## Reduce ##

> 1）接收合并好的数据，业务处理   
> 2）结果写入磁盘

## Output ##

> 输出reduce后的数据到文件系统中。

[mapreduce_.png-47kB]: /images/20220616/270c1fce14d341a1a4285b2be5bddea3.png
[c2.png-35.9kB]: /images/20220616/900b666064ba41babac5ab5106aa9724.png