初识MapReduce之这是个啥？

妖狐艹你老母 2024-04-18 11:42 61阅读 0赞

### 1. 是个啥 ###

###### 1.1 我觉得他是。。。 ######

**MapReduce是什么**  
MR是一个分布式运算程序程序编程框架,是用户开发"基于Hadoop的数据分析应用的"核心框架  
MR核心功能是将用户编写业务逻辑代码和自带的默认组件组合成一个完成的分布式运行程序,并运行在 一个Hadoop集群上  
**MR优点:**  
1.MR易于编程,它提供一些简单统一的接口,可以放程序猿完成一个分布式程序,这个分布式程序可以运行 在大量廉价的PC机器上.  
2.良好的扩展性:当计算资源不能满足的时候,可以通过简单的增加启动来扩展计算能力  
3.高容错性:假如一台机器挂了,它可以上面的计算任务转移到另外一个节点上,不会影响任务导致失败,这 个过程需要人工干涉,后期版本的Hadoop提供自动干预  
4.MR适合的是离线数据处理  
**MR缺点:**  
1.MR不适合实时计算,因为内部计算逻辑,所以MR不能以最快的速度得到结果  
2.MR不适合计算流式,不适合计算实时数据(动态的数据),MR适合计算静态数据(固定数据)  
3.MR中没有涉及一个逻辑DAG(有向无环图),每个MR作业的输出结果都会写到磁盘,就会造成大量的IO,会 导致性能下降  
**MR做什么**  
MapReduce 最主要的就是处理大数据它具备一个思想"分而治之"  
通过Mapper将分复杂的数据进行拆分,分解成若干个简单的任务来处理:  
1.数据或计算的规模会大大减少  
2.就近原则,将任务分别到最近的节点上进行计算  
3.因为将数据或计算进行拆分,会形成一些小的任务,可以并行计算,计算的时候彼此之间没有任何依赖关系  
Reducer负责将Mapper端处理好的数据进行汇总,Reducer(归约,合并)

###### 1.2 我觉得他应该是。。。。。 ######

分配记录车辆的个数,这个就是Map 每个人负责一个部分的车辆个数, 每个人将记录的车辆个数 提交给分配任务的人,此时会汇总得到数据Reduce ps:(原语)相同的key为一组,调用一次reduce进行计算  
在MapReduce中 其中Map不是我们印象中Map(HashMap),HashMap时候,因为有Hash表存在,所以 HashMap的key是排重,MR中Map是不会排重重复的它就是一个单纯的K,V键值对  
注意：map在读取文件时是按行读取，每一行都是一个map，过程如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2NTI1OTA2_size_16_color_FFFFFF_t_70]  
以wordcount为例，他的需求是这样似的：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2NTI1OTA2_size_16_color_FFFFFF_t_70 1]  
要求我们将一个文件中的单词做词频统计：  
最后出来的应该是键值对的形式的：Bye 3， Hadoop 4.。。。。。

按照标准的mapreduce过程他应该是：  
Map过程：并行读取三行，对读取的单词进行map操作，每个词都以<key,value>形式生成  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2NTI1OTA2_size_16_color_FFFFFF_t_70 2]  
shuffle过程：（溢写和排序），在map将数据规整好以后会将map的结果输出到outputcollector中，再由collector输出到环形缓冲区中，默认缓冲区大小为100M，默认超过80%即溢写到磁盘，将数据写到分区（默认一个分区），然后调用快排进行分区排序，最后再归并，形成K，V\[1,1,1,1\]形式的键值对（调用外部排序，归并排序）保证数据按key有序，最后进入reducetask  
或者这么说：  
先对数据进行切片,然后将数据传递给map,map的输出是内存缓冲区(圆形缓冲区),内存缓冲区默认大小 是100M,当 达到80%即0.8的时候将数据溢写到本地,剩余20%用于继续获取数据,在溢写到磁盘的时候会 执行partition(分区)和 sort(排序),然后对文件进行合并操作,合并完成之后 reducetask会启动线程去 mapTask拉取数据,然后进行文件合并, 并进行排序(归并),然后将小文件合并成一个大文件并调用reduce 方法完成最终输出效果

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

reduce操作是对map的结果进行排序，合并，最后得出词频：  
例如：Key：Bye，Value（1+1+1），即为<Bye,3>这样的键值对  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2NTI1OTA2_size_16_color_FFFFFF_t_70 4]

### 2 还有啥 ###

###### 2.1 还有一些术语需要了解一下 ######

1.  job作业：客户端需要执行的一个工作单元，包括输入数据，MR程序和配置信息
2.  task任务：Hadoop将job分成若干任务，包括map task和reduce task两类任务
    
    任务运行在集群节点上，通过YARN来调度， 如果一个任务失败，会在另一个节点上自动重新调度运行
3.  input split输入分片/分片：将MR的输入数据划分成等长的小数据块，每一块都是一个input split，Hadoop为每个分片创建一个map任务，该任务运行定义的map函数处理分片的每条记录
4.  数据本地化：Hadoop在存储有输入数据的节点上运行map任务，可以获得最佳性能，因为无需集群带宽资源，这就是“数据本地化优势”
5.  partition分区：如果reduce任务有多个，map任务可以针对输出进行分区(patititon)，为每个reduce任务创建一个分区，分区有多个键，每个键对应的键值记录都在同一个分区。分区可以由分区函数(默认为Hash函数)分区
6.  shuffle混洗：map任务和reduce任务之间的数据流成为shuffle，调整混洗参数对作业运行时间影响很大
7.  无reduce任务：当map任务完全并行时，可能无需reduce任务，唯一的非本地节点传输是map任务结果输出到HDFS
8.  combiner函数：为尽量避免map任务和reduce任务的数据传输，为map任务指定combiner函数，将combiner函数输出作为reducer任务输入。combiner做中间处理，但不能影响reduce任务的结果。在Java MR中，由Reducer类定义combiner，Job设置serCombinerClass为对应的Reducer类

分片的作用

1.  并行处理分片，降低处理时间
2.  切片被分的越细，负载均衡的质量越高
3.  如果分片太小，可能管理分片的时间和创建map任务的时间则会影响作业的执行时间，合理的分片大小等于HDFS块大小，默认128M，可以避免不同节点/机架/数据中心数据的传输

### 3 干点啥 ###

（举个栗子呀，记好格式）

package MR;
    
    import java.io.IOException;
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Job;
    import org.apache.hadoop.mapreduce.Mapper;
    import org.apache.hadoop.mapreduce.Reducer;
    import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    
    public class WC{
        
    //定义Mapper类，四个参数分别是（map读入的偏移量类型，值类型，map即将输出的键类型值类型）
        public static  class MyMap extends Mapper<LongWritable, Text, Text, IntWritable>{
        
            @Override
            protected void map(LongWritable key, Text value,Context context)
                    throws IOException, InterruptedException {
        
                String[] fields = value.toString().split(" ");//把读入的一行数据，按空格切割
                for(String s: fields){
        
                    context.write(new Text(s), new IntWritable(1));//每一单词都作为键值，并赋予value1，作为词频的基本数
                }
            }
        }
    
    //定义reducer类，四个参数分别是（map输出的键类型，值类型，reducer即将输出的键类型值类型）
        public static class MyReduce extends Reducer<Text, IntWritable, Text, IntWritable>{
        
            @Override
            protected void reduce(Text key, Iterable<IntWritable> values,//此时经过shuffle已经将value变为数组
                                  Context context) throws IOException, InterruptedException {
        
                int count=0;
                for(IntWritable i:values){
        
                    count+=i.get();//在ruducer里把每个值相加
                }
                context.write(key, new IntWritable(count));//
            }
        }
        public static void main(String[] args) {
        
            Configuration conf=new Configuration();
            try {
        
                Job job=Job.getInstance(conf);
                job.setJarByClass(WC.class);//设置主类
                job.setMapperClass(MyMap.class);//加载map类
                job.setReducerClass(MyReduce.class);//加载reduce类
    
                job.setOutputKeyClass(Text.class);//输出的KEY格式
                job.setOutputValueClass(IntWritable.class);//输出的value格式
    
                Path inPath =new Path("/input/words.txt");
                FileInputFormat.addInputPath(job, inPath);
    
                Path outpath=new Path("/output/WordCount/result");
                if(outpath.getFileSystem(conf).exists(outpath)){
        
                    outpath.getFileSystem(conf).delete(outpath, true);
                }
                FileOutputFormat.setOutputPath(job, outpath);
    
                job.waitForCompletion(true);
            } catch (Exception e) {
        
                // TODO Auto-generate
                //  .d catch block
                e.printStackTrace();
            }
        }
    }

迈过这个门槛，就能写一些mapreduce的应用了，其中combine、partition、grouping都会在以后的应用中用到。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2NTI1OTA2_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/88f0bbe579cb40c988230912ae766527.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2NTI1OTA2_size_16_color_FFFFFF_t_70 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/1029d6c02bb8421c8cc27bd63c20ac82.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2NTI1OTA2_size_16_color_FFFFFF_t_70 2]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/42c411967a684885900207ec476b2713.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2NTI1OTA2_size_16_color_FFFFFF_t_70 3]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/f3b34d69716f42c082d351f7aa61ece7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2NTI1OTA2_size_16_color_FFFFFF_t_70 4]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/031d37677c83436187d553af9f0cfd53.png