0. 引言
Hive中的常用算子包括distinct、join、group by、order by、distribute by、sort by、count等,这些操作符在SQL中使用起来很方便,能快速达到我们想要的效果,但是这些算子在底层是怎么实现的呢?
order by很容易想到执行原理,在一个reduce中将所有记录按值排序即可。因此order by在数据量大的情况下执行时间非常长,容易out of memory,非特殊业务需求一般不使用。distribute by也比较明显,根据hash值将distribute的值分发到不同的reduce。sort by是小号的order by,只负责将本reducer中的值排序,达到局部有序的效果。sort by和distribute by配合使用风味更佳,二者可以合并简写为cluster by。count则更加明晰,在combiner或reducer处按相同键累加值就能得到。
比较复杂的是distinct、join、group by,本文重点讨论这三个算子在MapReduce引擎中的大致实现原理。班门弄斧,抛砖引玉。
1. group by
map阶段,将group by后的字段组合作为key,如果group by单字段那么key就一个。将group by之后要进行的聚合操作字段作为值,如要进行count,则value是1;如要sum另一个字段,则value就是该字段。
shuffle阶段,按照key的不同分发到不同的reducer。注意此时可能因为key分布不均匀而出现数据倾斜的问题。
reduce阶段,将相同key的值累加或作其他需要的聚合操作,得到结果。
对group by的过程讲解的比较清楚的是这篇文章 http://www.mamicode.com/info-detail-2292193.html 图文并茂,很生动。
实例如下图,对应语句是 select rank, isonline, count(*) from city group by rank, isonline;
group by过程示意图如果group by出现数据倾斜,除去替换key为随机数、提前挑出大数量级key值等通用调优方法,适用于group by的特殊方法有以下几种:
(1)set hive.map.aggr=true,即开启map端的combiner,减少传到reducer的数据量,同时需设置参数hive.groupby.mapaggr.checkinterval 规定在 map 端进行聚合操作的条目数目。
(2)设置mapred.reduce.tasks为较大数量,降低每个reducer处理的数据量。
(3)set hive.groupby.skewindata=true,该参数可自动进行负载均衡。生成的查询计划会有两个 MR Job。第一个 MR Job 中,Map 的输出结果集合会随机分布到Reduce 中,每个 Reduce 做部分聚合操作,并输出结果,这样处理的结果是相同的 Group By Key有可能被分发到不同的 Reduce 中,从而达到负载均衡的目的;第二个 MR Job 再根据预处理的数据结果按照 Group ByKey 分布到 Reduce 中(这个过程可以保证相同的 Group By Key 被分布到同一个 Reduce中),最后完成最终的聚合操作。
2. join
Hive中有两种join方式:map join和common join
2.1 common join
如果不显式指定map side join,或者没有达到触发自动map join的条件,那么会进行reduce端的join,即common join,这种join包含map、shuffle、reduce三个步骤。
(1)Map阶段
读取源表的数据,Map输出时候以Join on条件中的列为key,如果Join有多个关联键,则以这些关联键的组合作为key。Map输出的value为join之后所关心的(select或者where中需要用到的)列;同时在value中还会包含表的Tag信息,用于标明此value对应哪个表。然后按照key进行排序。
(2)Shuffle阶段
根据key的值进行hash,并将key/value按照hash值推送至不同的reduce中,这样确保两个表中相同的key位于同一个reduce中
(3)Reduce阶段
根据key的值完成join操作,期间通过Tag来识别不同表中的数据。
以下面的SQL为例,可用下图所示过程大致表达其join原理。
SELECT u.name, o.orderid FROM user u JOIN order o ON u.uid = o.uid;
关联字段是uid,因此以uid为map阶段的输出key,value为选取的字段name和标记源表的tag。shuffle阶段将相同key的键值对发到一起,reduce阶段将不同源表、同一key值的记录拼接起来,可能存在一对多的情况。
common join原理示意2.2 map join
如果指定使用map join的方式,或者join的其中一张表小于某个体积(默认25MB),则会使用map join来执行。具体小表有多小,由参数hive.mapjoin.smalltable.filesize来决定。
Hive0.7之前,需要使用hint提示 /*+ mapjoin(table) */才会执行MapJoin,否则执行Common Join,但在0.7版本之后,默认自动会转换Map Join,由参数hive.auto.convert.join来控制,默认为true。
以下图为例说明map join如何执行,该图来自http://lxw1234.com/archives/2015/06/313.htm,博主是一个水平深厚又乐于分享的前辈,图片水印上也有其网址。
yarn会启动一个Local Task(在客户端本地执行的Task)--Task A,负责扫描小表b的数据,将其转换成一个HashTable的数据结构,并写入本地的文件中,之后将该文件加载到DistributeCache中。
接下来是Task B,该任务是一个没有Reduce的MR,启动MapTasks扫描大表a,在Map阶段,根据a的每一条记录去和DistributeCache中b表对应的HashTable关联,并直接输出结果。
由于MapJoin没有Reduce,所以由Map直接输出结果文件,有多少个Map Task,就有多少个结果文件。
map join原理示意3. distinct
distinct一般和group by同时出现。
3.1 distinct单字段
当distinct一个字段时,将group by的字段和distinct的字段组合在一起作为map输出的key,value设置为1,同时将group by的字段定为分区键,这可以确保相同group by字段的记录都分到同一个reducer,并且map的输入天然就是按照组合key排好序的。根据分区键将记录分发到reduce端后,按顺序取出组合键中的distinct字段,这时distinct字段也是排好序的。依次遍历distinct字段,每找到一个不同值,计数器就自增1,即可得到count distinct结果。例如下面的SQL语句,过程可以下图示意。
distinct 单字段select dealid,count(distinct uid) as num from order group by dealid;
3.2 distinct多字段
我暂时没有理解这是怎么实现的,别人写的也没有看明白。有善良的学富五车的大佬指点一下吗?
网友评论