MapReduce运行机制详解 09

╰+哭是因爲堅強的太久メ 2022-03-06 15:05 202阅读 0赞

# 1. Map的运行机制以及map任务的并行度 #

## 1.1 图解 ##

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

## 1.2 详解 ##

整个Map阶段流程大体如上图所示。简单概述：inputFile通过split被逻辑切分为多个split文件，通过Record按行读取内容给map（用户自己实现的）进行处理，数据被map处理结束之后交给OutputCollector收集器，对其结果key进行分区（默认使用hash分区），然后写入buffer，每个map task都有一个内存缓冲区，存储着map的输出结果，当缓冲区快满的时候需要将缓冲区的数据以一个临时文件的方式存放到磁盘，当整个map task结束后再对磁盘中这个map task产生的所有临时文件做合并，生成最终的正式输出文件，然后等待reduce task来拉数据。  
详细步骤：

*  首先，读取数据组件InputFormat（默认TextInputFormat）会通过getSplits方法对输入目录中文件进行逻辑切片规划得到splits，有多少个split就对应启动多少个MapTask。split与block的对应关系默认是一对一。
 *  将输入文件切分为splits之后，由RecordReader对象（默认LineRecordReader）进行读取，以\\n作为分隔符，读取一行数据，返回<key，value>。Key表示每行首字符偏移值，value表示这一行文本内容。
 *  读取split返回<key,value>，进入用户自己继承的Mapper类中，执行用户重写的map函数。RecordReader读取一行这里调用一次。
 *  map逻辑完之后，将map的每条结果通过context.write进行collect数据收集。在collect中，会先对其进行分区处理，默认使用HashPartitioner。  
    MapReduce提供Partitioner接口，它的作用就是根据key或value及reduce的数量来决定当前的这对输出数据最终应该交由哪个reduce task处理。默认对key hash后再以reduce task数量取模。默认的取模方式只是为了平均reduce的处理能力，如果用户自己对Partitioner有需求，可以订制并设置到job上。
 *  接下来，会将数据写入内存，内存中这片区域叫做环形缓冲区，缓冲区的作用是批量收集map结果，减少磁盘IO的影响。我们的key/value对以及Partition的结果都会被写入缓冲区。当然写入之前，key与value值都会被序列化成字节数组。  
    环形缓冲区其实是一个数组，数组中存放着key、value的序列化数据和key、value的元数据信息，包括partition、key的起始位置、value的起始位置以及value的长度。环形结构是一个抽象概念。  
    缓冲区是有大小限制，默认是100MB。当map task的输出结果很多时，就可能会撑爆内存，所以需要在一定条件下将缓冲区中的数据临时写入磁盘，然后重新利用这块缓冲区。这个从内存往磁盘写数据的过程被称为Spill，中文可译为溢写。这个溢写是由单独线程来完成，不影响往缓冲区写map结果的线程。溢写线程启动时不应该阻止map的结果输出，所以整个缓冲区有个溢写的比例spill.percent。这个比例默认是0.8，也就是当缓冲区的数据已经达到阈值（buffer size \* spill percent = 100MB \* 0.8 = 80MB），溢写线程启动，锁定这80MB的内存，执行溢写过程。Map task的输出结果还可以往剩下的20MB内存中写，互不影响。
 *  当溢写线程启动后，需要对这80MB空间内的key做排序(Sort)。排序是MapReduce模型默认的行为，这里的排序也是对序列化的字节做的排序。
    
     *  如果job设置过Combiner，那么现在就是使用Combiner的时候了。将有相同key的key/value对的value加起来，减少溢写到磁盘的数据量。Combiner会优化MapReduce的中间结果，所以它在整个模型中会多次使用。  
        那哪些场景才能使用Combiner呢？从这里分析，Combiner的输出是Reducer的输入，Combiner绝不能改变最终的计算结果。Combiner只应该用于那种Reduce的输入key/value与输出key/value类型完全一致，且不影响最终结果的场景。比如累加，最大值等。Combiner的使用一定得慎重，如果用好，它对job执行效率有帮助，反之会影响reduce的最终结果。
 *  合并溢写文件：每次溢写会在磁盘上生成一个临时文件（写之前判断是否有combiner），如果map的输出结果真的很大，有多次这样的溢写发生，磁盘上相应的就会有多个临时文件存在。当整个数据处理结束之后开始对磁盘中的临时文件进行merge合并，因为最终的文件只有一个，写入磁盘，并且为这个文件提供了一个索引文件，以记录每个reduce对应数据的偏移量。  
    至此map整个阶段结束。

## 1.3 基础设置 ##

mapTask的一些基础设置配置（mapred-site.xml当中社会）：  
设置一：设置环型缓冲区的内存值大小（默认设置如下）

mapreduce.task.io.sort.mb	100

设置二：设置溢写百分比（默认设置如下）

mapreduce.map.sort.spill.percent	0.80

设置三：设置溢写数据目录（默认设置）

mapreduce.cluster.local.dir	${hadoop.tmp.dir}/mapred/local

设置四：设置一次最多合并多少个溢写文件（默认设置如下）

mapreduce.task.io.sort.factor	10

# 2. ReduceTask 工作机制以及reduceTask的并行度 #

## 2.1 图解 ##

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

## 2.2 详解 ##

Reduce大致分为copy、sort、reduce三个阶段，重点在前两个阶段。copy阶段包含一个eventFetcher来获取已完成的map列表，由Fetcher线程去copy数据，在此过程中会启动两个merge线程，分别为inMemoryMerger和onDiskMerger，分别将内存中的数据merge到磁盘和将磁盘中的数据进行merge。待数据copy完成之后，copy阶段就完成了，开始进行sort阶段，sort阶段主要是执行finalMerge操作，纯粹的sort阶段，完成之后就是reduce阶段，调用用户定义的reduce函数进行处理。  
详细步骤：

*  Copy阶段，简单地拉取数据。Reduce进程启动一些数据copy线程(Fetcher)，通过HTTP方式请求maptask获取属于自己的文件。
 *  Merge阶段。这里的merge如map端的merge动作，只是数组中存放的是不同map端copy来的数值。Copy过来的数据会先放入内存缓冲区中，这里的缓冲区大小要比map端的更为灵活。merge有三种形式：内存到内存；内存到磁盘；磁盘到磁盘。默认情况下第一种形式不启用。当内存中的数据量到达一定阈值，就启动内存到磁盘的merge。与map 端类似，这也是溢写的过程，这个过程中如果你设置有Combiner，也是会启用的，然后在磁盘中生成了众多的溢写文件。第二种merge方式一直在运行，直到没有map端的数据时才结束，然后启动第三种磁盘到磁盘的merge方式生成最终的文件。
 *  合并排序。把分散的数据合并成一个大的数据后，还会再对合并后的数据排序。
 *  对排序后的键值对调用reduce方法，键相等的键值对调用一次reduce方法，每次调用会产生零个或者多个键值对，最后把这些输出的键值对写入到HDFS文件中。

# 3. MapReduce整个流程 #

## 3.1 清明上河图 ##

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

## 3.2 详解 ##

1、maptask的个数：取决于block块的个数  
2、reducetask的个数：job.setNumReduceTasks(2)  
3、环形缓冲区大小：100M  
4、溢写的比例：80%  
5、排序：两个地方都有排序  
\- map阶段的数据排序：局部的排序，每个maptask内部的数据会进行排序  
\- reduce阶段的数据排序：全局的排序，针对每个reudcetask的内部 的排序

# 4. MapReduce注意事项 #

*  map中有shuffle阶段的3个步骤, 环形缓存区 默认100M, 80%会触发溢写程序, 从环形缓存区(内存)–>内存或磁盘或内存+磁盘
 *  maptask的个数取决于TextInputFormat的个数, 取决于inputSplit的个数,
    
     *  inputSplit的大小取决于 mapred.min.split.size & mapred.max.split.size 的配置,如果不配置, inputSplit的大小默认为128M, 也就是一个block的大小
     *  所以,默认情况下 有多少个block,就要多少个maptask
 *  reduceTask 的个数 取决于job.setReduceNumTasks()
    
     *  默认reduceNum为1, 使用的是HasPartitioner 实现了Partitioner
        
         *  reduceNumTask的个数不会小于分区的个数
     *  如果是手动实现Partitioner, 需要手动设置reduceNumTask的个数
        
         *  如果reduceNumTask个数小于分区个数 报错 10块砖 只有5个人搬
         *  如果reduceNumTask个数大于分区个数 产生空文件 10块砖 只有15个人搬
         *  如果reduceNumTask个数等于分区个数 正好 10块砖 只有10个人搬
 *  压缩机制
    
     *  map输出数据压缩
     *  reduce输出数据压缩 snappy appach天生不支持这种编译, 需要手动编译
 *  计数器 TaskCounter
 *  规约 基础reducer map阶段就合并相同k2的数据, 减少k2数据量的传输

# 5. Join操作 #

## 5.1 reduce端join算法实现（无法解决数据倾斜问题） ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70 3]  
这种方式中，join的操作是在reduce阶段完成，reduce端的处理压力太大，map节点的运算负载则很低，资源利用率不高，且在reduce阶段极易产生数据倾斜![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70 4]

*  定义我们的OrderBean

public class OrderJoinBean implements Writable { 
        private String id;
        private String date;
        private String pid;
        private String amount;
        private String name;
        private String categoryId;
        private String price;
        @Override
        public String toString() { 
            return id+"\t"+date+"\t"+pid+"\t"+amount+"\t"+name+"\t"+categoryId+"\t"+price;
        }
        public OrderJoinBean() { 
        }
    
        public OrderJoinBean(String id, String date, String pid, String amount, String name, String categoryId, String price) { 
            this.id = id;
            this.date = date;
            this.pid = pid;
            this.amount = amount;
            this.name = name;
            this.categoryId = categoryId;
            this.price = price;
        }
        public String getId() { 
            return id;
        }
    
        public void setId(String id) { 
            this.id = id;
        }
    
        public String getDate() { 
            return date;
        }
        public void setDate(String date) { 
            this.date = date;
        }
        public String getPid() { 
            return pid;
        }
        public void setPid(String pid) { 
            this.pid = pid;
        }
        public String getAmount() { 
            return amount;
        }
        public void setAmount(String amount) { 
            this.amount = amount;
        }
        public String getName() { 
            return name;
        }
        public void setName(String name) { 
            this.name = name;
        }
        public String getCategoryId() { 
            return categoryId;
        }
        public void setCategoryId(String categoryId) { 
            this.categoryId = categoryId;
        }
        public String getPrice() { 
            return price;
        }
    
        public void setPrice(String price) { 
            this.price = price;
        }
    
        @Override
        public void write(DataOutput out) throws IOException { 
            out.writeUTF(id+"");
            out.writeUTF(date+"");
            out.writeUTF(pid+"");
            out.writeUTF(amount+"");
            out.writeUTF(name+"");
            out.writeUTF(categoryId+"");
            out.writeUTF(price+"");
        }
    
        @Override
        public void readFields(DataInput in) throws IOException { 
           this.id =  in.readUTF();
            this.date =  in.readUTF();
            this.pid =  in.readUTF();
            this.amount =  in.readUTF();
            this.name =  in.readUTF();
            this.categoryId =  in.readUTF();
            this.price =  in.readUTF();
    
        }
    }

*  定义我们的map类

public class OrderJoinMap extends Mapper<LongWritable,Text,Text,OrderJoinBean> { 
        private OrderJoinBean orderJoinBean = new OrderJoinBean();
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { 
           //通过获取文件名来区分两个不同的文件
            String[] split = value.toString().split(",");
            FileSplit inputSplit = (FileSplit) context.getInputSplit();
            String name = inputSplit.getPath().getName();
            System.out.println("获取文件名为"+name);
            if(name.contains("orders")){ 
                //订单数据
                orderJoinBean.setId(split[0]);
                orderJoinBean.setDate(split[1]);
                orderJoinBean.setPid(split[2]);
                orderJoinBean.setAmount(split[3]);
                context.write(new Text(split[2]),orderJoinBean);
            }else{ 
                //商品数据
                orderJoinBean.setName(split[1]);
                orderJoinBean.setCategoryId(split[2]);
                orderJoinBean.setPrice(split[3]);
                context.write(new Text(split[0]),orderJoinBean);
            }
    
        }
    }

*  定义我们的reduce类

public class OrderJoinReduce extends Reducer<Text,OrderJoinBean,OrderJoinBean,NullWritable> { 
        private OrderJoinBean orderJoinBean;
        @Override
        protected void reduce(Text key, Iterable<OrderJoinBean> values, Context context) throws IOException, InterruptedException { 
             orderJoinBean = new OrderJoinBean();
            for (OrderJoinBean value : values) { 
                System.out.println(value.getId());
                //相同的key的对象都发送到了这里，在这里将数据拼接完整
              if(null !=value.getId() && !value.getId().equals("null") ){ 
                  orderJoinBean.setId(value.getId());
                  orderJoinBean.setDate(value.getDate());
                  orderJoinBean.setPid(value.getPid());
                  orderJoinBean.setAmount(value.getAmount());
              }else{ 
                  orderJoinBean.setName(value.getName());
                  orderJoinBean.setCategoryId(value.getCategoryId());
                  orderJoinBean.setPrice(value.getPrice());
              }
            }
            context.write(orderJoinBean,NullWritable.get());
        }
    }

*  开发main方法入口

public class OrderJoinMain extends Configured implements Tool { 
        @Override
        public int run(String[] args) throws Exception { 
            Job job = Job.getInstance(super.getConf(), OrderJoinMain.class.getSimpleName());
    
            job.setInputFormatClass(TextInputFormat.class);
            TextInputFormat.addInputPath(job,new Path("file:///F:\\传智播客大数据离线阶段课程资料\\4、大数据离线第四天\\map端join\\input"));
            job.setMapperClass(OrderJoinMap.class);
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(OrderJoinBean.class);
            job.setReducerClass(OrderJoinReduce.class);
            job.setOutputKeyClass(OrderJoinBean.class);
            job.setOutputValueClass(NullWritable.class);
            job.setOutputFormatClass(TextOutputFormat.class);
            TextOutputFormat.setOutputPath(job,new Path("file:///F:\\传智播客大数据离线阶段课程资料\\4、大数据离线第四天\\map端join\\out"));
            boolean b = job.waitForCompletion(true);
            return b?0:1;
        }
    
        public static void main(String[] args) throws Exception { 
            ToolRunner.run(new Configuration(),new OrderJoinMain(),args);
        }
    }

## 5.2 map端join算法实现 ##

### 5.2.1 原理及示例 ###

*  原理阐述  
    适用于关联表中有小表的情形；  
    可以将小表分发到所有的map节点，这样，map节点就可以在本地对自己所读到的大表数据进行join并输出最终结果，可以大大提高join操作的并发度，加快处理速度
 *  实现示例  
    先在mapper类中预先定义好小表，进行join  
    引入实际场景中的解决方案：一次加载数据库或者用  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70 5]

### 5.2.2 代码实现 ###

*  定义mapJoin

public class JoinMap extends Mapper<LongWritable,Text,Text,Text> { 
        HashMap<String,String> b_tab = new HashMap<String, String>();
        String line = null;
        /* map端的初始化方法当中获取我们的缓存文件，一次性加载到map当中来 */
        @Override
        public void setup(Context context) throws IOException, InterruptedException { 
            //这种方式获取所有的缓存文件
         // URI[] cacheFiles1 = DistributedCache.getCacheFiles(context.getConfiguration());
            Path[] localCacheFiles = DistributedCache.getLocalCacheFiles(context.getConfiguration());
            URI[] cacheFiles = DistributedCache.getCacheFiles(context.getConfiguration());
            FileSystem fileSystem = FileSystem.get(cacheFiles[0], context.getConfiguration());
            FSDataInputStream open = fileSystem.open(new Path(cacheFiles[0]));
            BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(open));
            while ((line = bufferedReader.readLine())!=null){ 
                String[] split = line.split(",");
                b_tab.put(split[0],split[1]+"\t"+split[2]+"\t"+split[3]);
            }
            fileSystem.close();
            IOUtils.closeStream(bufferedReader);
        }
    
        @Override
        public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { 
            //这里读的是这个map task所负责的那一个切片数据（在hdfs上）
            String[] fields = value.toString().split(",");
            String orderId = fields[0];
            String date = fields[1];
            String pdId = fields[2];
            String amount = fields[3];
            //获取map当中的商品详细信息
            String productInfo = b_tab.get(pdId);
            context.write(new Text(orderId), new Text(date + "\t" + productInfo+"\t"+amount));
        }
    }

*  定义程序运行main方法

public class MapSideJoin extends Configured implements Tool { 
        @Override
        public int run(String[] args) throws Exception { 
            Configuration conf = super.getConf();
            //注意，这里的缓存文件的添加，只能将缓存文件放到hdfs文件系统当中，放到本地加载不到
            DistributedCache.addCacheFile(new URI("hdfs://192.168.52.100:8020/cachefile/pdts.txt"),conf);
            Job job = Job.getInstance(conf, MapSideJoin.class.getSimpleName());
            job.setJarByClass(MapSideJoin.class);
            job.setInputFormatClass(TextInputFormat.class);
            TextInputFormat.addInputPath(job,new Path("file:///F:\\传智播客大数据离线阶段课程资料\\4、大数据离线第四天\\map端join\\map_join_iput"));
            job.setMapperClass(JoinMap.class);
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(Text.class);
            job.setOutputFormatClass(TextOutputFormat.class);
            TextOutputFormat.setOutputPath(job,new Path("file:///F:\\传智播客大数据离线阶段课程资料\\4、大数据离线第四天\\map端join\\map_join_output")) ;
            boolean b = job.waitForCompletion(true);
            return b?0:1;
        }
        public static void main(String[] args) throws Exception { 
            Configuration configuration = new Configuration();
            ToolRunner.run(configuration,new MapSideJoin(),args);
    
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70]: /images/20220302/69d9f33a6b574c4eb125ec3dcd7ea3b9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70 1]: /images/20220302/674d10f9f901441fb10943969287ef8b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70 2]: /images/20220302/61945f0f19554e3c9d1beacbb6b94191.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70 3]: /images/20220302/11e4e5b1751d4789972f20cbd9ef1fb4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70 4]: /images/20220302/c654e67f992e4cbeba1d93b6bf5cc045.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70 5]: /images/20220302/4feb877833654d11ad21d608296ae7b0.png