3.1mapreduce的shuffle机制

3.1.1 概述：

v mapreduce中，map阶段处理的数据如何传递给reduce阶段，是mapreduce框架中最关键的一个流程，这个流程就叫shuffle；

让客户满意是我们工作的目标，不断超越客户的期望值来自于我们对这个行业的热爱。我们立志把好的技术通过有效、简单的方式提供给客户，将通过不懈努力成为客户在信息化领域值得信任、有价值的长期合作伙伴，公司提供的服务项目有：域名与空间、虚拟主机、营销软件、网站建设、耀州网站维护、网站推广。

v shuffle: 洗牌、发牌——（核心机制：数据分区，排序，缓存）；

v 具体来说：就是将maptask输出的处理结果数据，分发给reducetask，并在分发的过程中，对数据按key进行了分区和排序；

 

3.1.2 主要流程：

Shuffle缓存流程：

 

shuffle是MR处理流程中的一个过程，它的每一个处理步骤是分散在各个map task和reduce task节点上完成的，整体来看，分为3个操作：

1、分区partition

2、Sort根据key排序

3、Combiner进行局部value的合并

 

3.1.3 详细流程

1、 maptask收集我们的map()方法输出的kv对，放到内存缓冲区中

2、 从内存缓冲区不断溢出本地磁盘文件，可能会溢出多个文件

3、 多个溢出文件会被合并成大的溢出文件

4、 在溢出过程中，及合并的过程中，都要调用partitoner进行分组和针对key进行排序

5、 reducetask根据自己的分区号，去各个maptask机器上取相应的结果分区数据

6、 reducetask会取到同一个分区的来自不同maptask的结果文件，reducetask会将这些文件再进行合并（归并排序）

7、 合并成大文件后，shuffle的过程也就结束了，后面进入reducetask的逻辑运算过程（从文件中取出一个一个的键值对group，调用用户自定义的reduce()方法）

 

Shuffle中的缓冲区大小会影响到mapreduce程序的执行效率，原则上说，缓冲区越大，磁盘io的次数越少，执行速度就越快

缓冲区的大小可以通过参数调整,  参数：io.sort.mb  默认100M

 

 

 

3.1.4 详细流程示意图

 

3.2. MAPREDUCE中的序列化

3.2.1 概述

Java的序列化是一个重量级序列化框架（Serializable），一个对象被序列化后，会附带很多额外的信息（各种校验信息，header，继承体系。。。。），不便于在网络中高效传输；

所以，hadoop自己开发了一套序列化机制（Writable），精简，高效

 

3.2.2Jdk序列化和MR序列化之间的比较

简单代码验证两种序列化机制的差别：

public class TestSeri { public static void main(String[] args) throws Exception { //定义两个ByteArrayOutputStream，用来接收不同序列化机制的序列化结果 ByteArrayOutputStream ba = new ByteArrayOutputStream(); ByteArrayOutputStream ba2 = new ByteArrayOutputStream();   //定义两个DataOutputStream，用于将普通对象进行jdk标准序列化 DataOutputStream dout = new DataOutputStream(ba); DataOutputStream dout2 = new DataOutputStream(ba2); ObjectOutputStream obout = new ObjectOutputStream(dout2); //定义两个bean，作为序列化的源对象 ItemBeanSer itemBeanSer = new ItemBeanSer(1000L, 89.9f); ItemBean itemBean = new ItemBean(1000L, 89.9f);   //用于比较String类型和Text类型的序列化差别 Text atext = new Text("a"); // atext.write(dout); itemBean.write(dout);   byte[] byteArray = ba.toByteArray();   //比较序列化结果 System.out.println(byteArray.length); for (byte b : byteArray) {   System.out.print(b); System.out.print(":"); }   System.out.println("-----------------------");   String astr = "a"; // dout2.writeUTF(astr); obout.writeObject(itemBeanSer);   byte[] byteArray2 = ba2.toByteArray(); System.out.println(byteArray2.length); for (byte b : byteArray2) { System.out.print(b); System.out.print(":"); } } }

 

 

 

3.2.3 自定义对象实现MR中的序列化接口

如果需要将自定义的bean放在key中传输，则还需要实现comparable接口，因为mapreduce框中的shuffle过程一定会对key进行排序,此时，自定义的bean实现的接口应该是：

public class FlowBean implements WritableComparable

需要自己实现的方法是：

/**  * 反序列化的方法，反序列化时，从流中读取到的各个字段的顺序应该与序列化时写出去的顺序保持一致  */ @Override public void readFields(DataInput in) throws IOException { upflow = in.readLong(); dflow = in.readLong(); sumflow = in.readLong();   }   /**  * 序列化的方法  */ @Override public void write(DataOutput out) throws IOException {   out.writeLong(upflow); out.writeLong(dflow); //可以考虑不序列化总流量，因为总流量是可以通过上行流量和下行流量计算出来的 out.writeLong(sumflow);   } @Override public int compareTo(FlowBean o) { //实现按照sumflow的大小倒序排序 return sumflow>o.getSumflow()?-1:1; }

 

 

 

3.3. MapReduce与YARN

3.3.1 YARN概述

Yarn是一个资源调度平台，负责为运算程序提供服务器运算资源，相当于一个分布式的操作系统平台，而mapreduce等运算程序则相当于运行于操作系统之上的应用程序

3.3.2 YARN的重要概念

1、 yarn并不清楚用户提交的程序的运行机制

2、 yarn只提供运算资源的调度（用户程序向yarn申请资源，yarn就负责分配资源）

3、 yarn中的主管角色叫ResourceManager

4、 yarn中具体提供运算资源的角色叫NodeManager

5、 这样一来，yarn其实就与运行的用户程序完全解耦，就意味着yarn上可以运行各种类型的分布式运算程序（mapreduce只是其中的一种），比如mapreduce、storm程序，spark程序，tez……

6、 所以，spark、storm等运算框架都可以整合在yarn上运行，只要他们各自的框架中有符合yarn规范的资源请求机制即可

7、 Yarn就成为一个通用的资源调度平台，从此，企业中以前存在的各种运算集群都可以整合在一个物理集群上，提高资源利用率，方便数据共享

 

3.3.3Yarn中运行运算程序的示例

mapreduce程序的调度过程，如下图

 

 

文章名称：MAPREDUCE原理篇（2）
本文地址：http://kswjz.com/article/gpcejs.html