Mapreduce与Hive比较

网友 CSDN博客

1、背景介绍

早先的工作中,有很多比较复杂的分析工作,当时对hive还不熟悉,但是java比较熟悉,所以在进行处理的时候,优先选择了MR.

但是随着工作的数据内容越来越多,越来越复杂,对应的调整也越来越多,越来越复杂.纯使用MR方式整个流程就比较复杂,如果需要修改某个部分,那首先需要修改代码中的逻辑,然后把代码打包上传到某个可访问路径上(一般就是hdfs),然后在调度平台内执行.如果改动较大的情况,可能还会需要在测试环境中多次调试. 总之就是会花比较多的时间在非业务逻辑改动的工作上.

考虑到维护的成本的增大,慢慢的开始准备将MR的作业,逐渐的移植到一些脚本平台上去,hive成了我们的首选。

2、实战场景

先看这样一个场景. 每一个用户在登录到网站上的时候会带有一个ip地址,多次登录可能会有多个不同的ip地址.

假设,我们已经有一个用户->ip地址这样的一份数据.我们需要对此进行分析,得到一份来自相同ip的用户的关系表,数据格式类似

用户->用户,具体的ip我们不保留了。

第1步用udf取最频繁ip

我们先看一下原始数据的字段,是user_id,ips,我们再来看看ips内容的格式,我们执行

Select * from iptable limit 100

你会发现,虽然我们limit了100而且是没有任何复杂条件的查询,hive竟然也会去扫描所有的数据,这非常奇怪也很浪费。原来hive的limit在默认的情况下的执行过程就是把所有数据都跑出来,然后再一个reduce上,进行limit。这是为了保证在某些情况下筛选条件对结果的影响。

但是我们可以通过打开一个hive.limit.optimize.enable=true来简化这个查询,当这个选项打开以后hive会读取hive.limit.row.max.size,hive.limit.optimize.limit.file的默认值来进行小数据量的计算。

我们看到ips的原始数据的格式是ip,ip,… 用逗号分隔的多个ip字符串。

我们要从用户->[ip地址] 这样的数据中得到一个用户使用最多的ip地址作为用户的最常用ip。这里我们会使用hive的自定义udf来完成这一步的工作。

那么udf是什么呢,udf就是user define function的缩写.通过它我们可以对hive进行扩展,hive本身已经带了很多的基本的udf了,比如length(),sin(),unix_timestamp(),regexp_replace()等等.

这些都是一些比较通用的处理,如果有的时候我们要在字段上做一些特殊的逻辑就要自己动手写了.

下面就是我们用来实现这个功能的udf代码

@Description(name = “freq

ips”, value = “find most frequence ips from all login ip”, extended = “”)

public class FindFreqIps extends UDF {

public String evaluate(String content, int limit) {

// 计算最常用ip的代码逻辑,并返回结果

Return result;

}

里面的逻辑主要就是找到前limit个最长使用的Ip,我们看到我们的类需要继承自hive包中的UDF类,然后任意的定义输入类型和返回类型,但是方法的名字一定要叫evaluate,hive会使用反射来得到这个方法的输入输出。当我们要在hive中使用它的时候,我们要首先把这个类打成jar包,然后让hive可以访问到。一般可以直接放在hdfs上,然后使用

Add jar hdfs_path/myjar.jar;

Create temporary function FindFreqIps as ‘FindFreqIps’

Select user_id, FindFreqIps(ips) as freqIps from tablexxx

另外还有一种是继承自genericUDF,这种方式可以自由的控制输入和返回类型处理,比起UDF来说更加的灵活些。但是我们这里普通的udf就足够了。

第2步列转行,进行join

从第一步,我们得到了用户最常用的N个ip,我们这里假设值3个。然后我们要找到这些用户之间的关联,即相同的ip的关系。

那么非常直接的方式,我们直接对用户的ip进行join,但是现在ip是3个连在一起字符串的形式,无法直接join。那么我们就先把ip都分解开。

我们把这个ips的字段进行一个列转行的转换,如下

Select user_id,ip from tablexxx

Lateral view explode(split(ips, “,”))

subview as ip

这样就会得到 user->ip的单条的记录。这里的

这下要join就方便了,假设上面的结果表是singleIP我们

Select a.user_id as fromid, b.user_id as

toid

SingleIP a

Join SingleIP b

On a.ip = b.ip and a.user_id <>

b.user_id;

什么,报错了!

a.user_id <> b.user_id这个部分会报错,因为hive中join的时候,是只能指定等式来进行匹配的,不支持不等式的条件。如果使用了不等式,会使join的数量变的非常大。

于是,我们就只能曲线救国了。

Select * from

(Select a.user_id as fromid, b.user_id as

toid

SingleIP a Join

SingleIP b

On a.ip = b.ip) m

Where m.fromid <> m.toid;

你会发现,执行了1次join,2次select使用的mr的步骤还是一步。一般总感觉嵌套了一次select以后也会对应的产生2次mr,难道是hive自己进行了优化吗?那么我们借助hive的分析工具来看看hive是如何执行的呢。

我们在刚才的语句前加上explain,来看看这个select的执行计划。

Hive会通过antlr来对输入的sql语句进行语法分析,产生一个执行计划。

执行计划会有三个部分

第一部分是ABSTRACT SYNTAX

TREE抽象语法树

这里面显示了hive把这个sql解析成什么样的各个token。

类似这样(TOK_SELECT (TOK_SELEXPR TOK_ALLCOLREF))表示

第二部分是STAGE DEPENDENCIES

一个hive的执行过程会包含多个stage,他们之间有互相依赖的关系。比如下面

Stage-1 is a root

stage

Stage-0 depends on stages: Stage-1

Stage-3 depends on

stages: Stage-0

这里的stage-1是root stage。而0依赖于1,3依赖于0。

第三个部分是STAGE PLANS, 就是每个stage中的具体执行的步骤。

在stage plans里面每一个stage和普通的hadoop程序一样会有map和reduce的过程。我们截取一段map过程的计划看下。

Stage: Stage-1

Map Reduce

Alias ->

Map Operator Tree:

TableScan

alias:

Reduce

Output Operator

key expressions:

expr: ip

type: string

sort

order: +

Map-reduce partition columns:

expr: ip

type: string

tag:

value expressions:

expr: user_id

type: string

这里是对a表也就是SingleIP表的一个map阶段的操作。Reduce output operator这里会显示使用ip作为key,自增排序,因为是string的所以是字典序的自增。Partition使用ip作为分发字段。tag指的是类似一个来源的概念,因为这里的join采用的是reduce join的方式,每一个从不同的map来的数据最后在reduce进行汇合,他们会被打上一个标记,代表他们的来源。然后就是value的内容,user_id。

然后再来看看reduce过程的计划

Reduce Operator Tree:

Join

Operator

condition

map:

Inner Join 0 to 1

condition

expressions:

{VALUE._col0}

handleSkewJoin: false

outputColumnNames: _col0, _col2

这里显示一个join的操作。这里表示把0的内容加到1上。后面有一个handleSkewJoin,这个是hive的一个应对数据倾斜的一种处理方式,默认是关闭的,我们后面再来详细看。

这里也可以用explain extended,输出的信息会更加详细。那么看了这个我们再比较一下我们之前的第二个查询计划,我们来看看加上了嵌套查询以后的执行计划有什么变化呢?会发现hive在reduce的执行计划里面会加上一段

Filter

Operator

predicate:

expr: (_col0 <> _col2)

type: boolean

在reduce最后输出之前,进行了一个过滤的操作,过滤的条件就是外部的查询的where条件。正如我们所料,hive发现这个过程是可以一次性完成的,所以进行了优化,放在了reduce阶段来作了。

另外如果hive中有多张表进行join,如果他们的join key是一样的,那么hive就会把他们都放在一次mr中完成。

第3步数据倾斜

上一步中,我们计算出了所有的相同ip的人的点对点关系。但是这个结果集会有不少问题,比如如果某个ip是一个公共出口,那么就会出现同一个ip有上万人都在使用,他们互相join展开以后,结果的数据量会非常大,时间上很慢不说,最终得到的数据实际上很多我们也用不上(这个是基于业务上得考虑),甚至有可能,在展开的时候会出现各种问题,导致计算时间过长,算不出来。这种情况,我们在hive里面称之为数据倾斜。

在group by的时候,如果出现某一个reduce上得数据量过大的情况,hive有一个默认的hive.groupby.skewindata选项,当把它设置为true的时候,hive会将原来的一次MR变成2次,第一次,数据在reduce的时候会随机分发到每个reduce,做部分的聚合,然后第二次的时候再按照group by的key进行分发。这样可以有效的处理一般的倾斜情况。

而在join的时候,如果join的其中某个key的值非常的多,也会导致倾斜。有的时候,如果有null值,在hive看来null和null是相等的,它也会对他们进行join,也会错误的倾斜。由于join的时候,hive会把第一张表的内容放到一个内容map中,然后不断的读取后表的内容来进行join,所以如果左边的表示小表这个过程就会非常的高效。当然使用mapjoin也一种有效的方式,直接把一张足够小的表完全放到内存来后另一张表进行join。类似这样

SELECT /*+ MAPJOIN(b) */

a.key, a.value FROM a join b on a.key = b.key;

我们的ip计算使用的是自己join自己,所有也没有大小表之分,同时单表的数据量也大到无法完全放进内存,那么是不是就要进行硬算呢?在实际中,因为ip的分布没有倾斜到太过火的程度,硬算也确实可以,但是这里我们换一种方式来稍稍优化一下。

首先我们采用bucket的方式来保存之间的用户->ip的数据。使用ip来作为分桶键。

CREATE TABLE userip(user_id bigint, ip STRING)

CLUSTERED BY (ip) INTO 128 BUCKETS;

然后set hive.enforce.bucketing = true;开启bucket计算

from tableaaa

insert overwrite table tablebbb

select user_id, ip;

结果将会被保存到128个不同的桶中,默认根据ip的hashcode来取模。这样每个桶内的数据基本大概是原数据量的1/100。当然如果原始数据量太大,还可以分桶更加多一些。

这个地方如果我们不开启enforce.bucketing的话,也可以通过设置

set mapred.reduce.tasks=128.然后在查询中cluster by来强制指定进行分桶。这步完成之后,我们再来进行设置

set hive.optimize.bucketmapjoin=true;

set hive.optimize.bucketmapjoin.sortedmerge=true;

然后hive就能对每个分块数据进行mapjoin。

第4步用udaf取top N

好了,现在我们已经有所有的user->user的数据,我们希望要一个user->[users]的一对多的记录,但是这个数据量有点大,实际上每个用户大概关联1000个已经足够了。首先对数据进行排序,排序的依据就是按照用户的相同的ip的数量。然后去最前面的1000个,不足的按实际数量取。

这个地方比较容易想到的就是,先group by fromid,toid,然后count一个总数作为新字段,如下

这里想到一种做法是用淘宝的一个类sql的row_number实现,然后用row_number来对fromid做主键,给按照count从大到小写上序列编号seq。最后做一个嵌套查询,只取seq<=1000的数据。Row_number的话标准的hive中没有。那么这里就可以让自定义udaf上场了。

Udaf顾名思义就是一个Aggregate的udf,和之前的udf的区别就是他一般是用来group by的场合中。

@Description(name = “myudaf”, value = “calc users has most same ips ” )

public class GenericUDAFCollect extends AbstractGenericUDAFResolver {

@Override

public GenericUDAFEvaluator getEvaluator(TypeInfo[] parameters)

throws SemanticException {

return new MyUDAFEvaluator();

}

自己定义一个evaluator,并且实现其中的一些方法。

public static class MyUDAFEvaluator extends GenericUDAFEvaluator {

@Override

public ObjectInspector init(Mode m, ObjectInspector[] parameters) throws HiveException {

}

@Override

public void reset(AggregationBuffer agg) throws HiveException {

}

@Override

public AggregationBuffer getNewAggregationBuffer() throws HiveException {

}

// Mapside

@Override

public void iterate(AggregationBuffer agg, Object[] parameters) throws HiveException {

}

// Mapside

@Override

public Object terminatePartial(AggregationBuffer agg) throws HiveException {

}

@Override

public void merge(AggregationBuffer agg, Object partial) throws HiveException {

}

// Reduceside

@Override

public Object terminate(AggregationBuffer agg) throws HiveException {

}

在init阶段会传一个Mode进来,这个Mode中定义了以下的几个阶段

PARTIAL1: 这个是map阶段,这个阶段会调用iterate(),和terminatePartial()

PARTIAL2: 这个是map段得combiner阶段,会将map端的数据进行合并,也可能没有这个阶段。会执行merge()和terminatePartial()

FINAL: 这个是reduce阶段,会调用merge()和terminate()

COMPLETE: 这是纯map处理,无reduce的情况出现的阶段,它会调用iterate()和terminate()

而从函数方面来说,init是初始化,他会传入mode作为参数。可以根据不同的阶段采取不同的处理。getNewAggregationBuffer的处理是hive为了内存的复用,减少gc,他并不是每一次处理一条记录都会新申请空间,而是在处理一批数据的时候重复使用一批内存。Terminate就是最终的输出了。

Ok,了解了udaf,那么可以动手了。Sql如下

Select fromid, getTopN(toid,n) from tablexx3

Group by fromid

其中的getTopN首先在map端,将每一个fromid的关联的toid的次数都记录下来,记录条数代表重复的ip数量,然后按照这个次数进行倒序排序,截取前n个。

在reduce端,将各个map端的结果再按照次数倒序排序,再进行截取n个并进行合并。最终输出的就是每个fromid对应的toid的列表了。

从这次从mr转换到hive的过程中,对我们目前的mr和hive进行了一些比较

3、mr和hive比较

1. 运算资源消耗

无论从时间,数据量,计算量上来看,一般情况下mr都是优于或者等于hive的。mr的灵活性是毋庸置疑的。在转换到hive的过程中,会有一些为了实现某些场景的需求而不得不用多步hive来实现的时候。

2. 开发成本/维护成本

毫无疑问,hive的开发成本是远低于mr的。如果能熟练的运用udf和transform会更加提高hvie开发的效率。另外对于数据的操作也非常的直观,对于全世界程序员都喜闻乐见的sql语法的继承也让它更加的容易上手。

hive独有的分区管理,方便进行数据的管理。