Hive.分组排序和TOP

HQL作为类SQL的查询分析语言，到目前为止，应该也还未能达到其它流行的SQL（如Transact-SQL, MySQL）实现那样完善。而在公司的生产环境中，我想应该也不会紧贴Hive版本更新的步伐，始终部署最新版的Hive；可能会滞后一两个大版本神马的；毕竟，虽然开源工具的透明性是一大利好，但与闭源的商业工具相比，在可用性等问题上的保障性还是略弱。

使用HQL进行离线分析用户数据时，就算已经过聚合处理，但我们也可能只对那些突出的量化指标或者这些指标的增量变化感兴趣，所以对聚合数据排序（按某列降序？增序？）成为很基本的需要，这在HQL这样尚未成熟的语言中，结合orderby, limit子句可以毫无鸭梨地完成。

然而，即使我们可以把多个字段放入order by子句中，并指定各个字段的升降顺序，如：

[sql] view
plaincopyprint?





order by fieldA desc, fieldB [asc], fieldC desc

但排序操作始终是全局的，我们有时候想要的却是分组排序，即按fieldA排序以后，然后针对fieldA的每个值所对应的fieldB和（或）fieldC排序，而不是像order by那样，针对所有fieldA的值对fieldB和（或）fieldC排序。

为了满足这个需要，Transact-SQL提供了over, partition by句和 row_number()函数，而Hive也在0.11中引入over, partition by子句和rank函数，以此提供方便的窗口分析（分组分析）功能。

那对于0.11版之前的Hive，我们可以实现分组排序吗？答案是肯定的，只是看起来没那么直接。

要实现这个需求，就需要请出distribute by, sort by这两个重要角色了，distribute by能够执行我们需要的分组功能，再结合Hive查询的MapReduce Job特性，sort by又可以在分组内进行局部排序。

当然，如果只有它们，我们只能得到排序后的一堆数据，但是无法知道每一条数据的名次，这就要自己编写UDF函数，来确定和返回名次了，这个函数貌似在网络上流传甚广：

[java] view
plaincopyprint?





public final class Rank extends UDF {

private int counter;

private String last_key ="";

public int evaluate(final String key) {

if (key == null) {

this.last_key= "";

this.counter= 0;

return counter;

}

if(!key.equalsIgnoreCase(this.last_key)) {

this.counter= 0;

this.last_key= key;

}

return this.counter++;

}

}

在这里我们忽略了自定义UDF的注册的环节。。。在分组之后，应用Rank函数，这个函数始终跟踪最新的参数值，在参数值连续相同的情况下，就将字段counter作自增操作并返回这个计数值；而如果出现和上一次函数调用不同的参数值，Rank函数会重置其计数值字段和key字段（对应参数值）使我们得到一个int类型的名次值。

Hive里称这个为自定义函数，实际上每个自定义函数是一个继承了UDF类，并提供和实现了evaluate方法的类；这个叫法略不福啊。

有了distribute by, sort by和这个Rank函数，我们就能够实现分组排序了，编写HQL查询脚本之前，我们还需要明确：

1. 分组字段：distribute by的字段是哪个（些）？

2. 排序依据：sort by的字段是哪个（些）？

3. 函数参数：Rank函数需要String参数，我们应该给Rank函数传递什么东西作为实参？

不晓得是因为以上这3个问题确实像我们在教科书上偶尔会看到的“容易证得”，还是因为博主们只是想了这个问题，并没有实践，反正我看到的网络上讲分组排序（TOP）的博文都没有明确地提出这3个问题。而按我的实践经历来看，初次实现这种需求的童鞋，就算看了这些博文，在得到正确结果之前，应该都会经历各种困惑，下面我们从实际的场景来看看。

比如，我们需要查询得到这样的数据：每日使用应用myAPP的UV量TOP 30的设备，和这TOP 10的设备中每个设备的流量（VV）最高的10项版块内容（以内容ID来区分）;

假定查询所需的Hive表为：hiveTab_useraction

思路梗概：先用一个子查询查出来每台设备的访问内容，同时，用一个子查询查出来TOP30的设备，然后两个表做内连接（join），然后在外层查询中提取所需字段列和数据列。

在这个流程里面：

1）找出TOP10的设备这个环节看起来没有涉及分组排序，但还是需要考虑上面3个问题，因为我们要得到名次，而order by貌似不能同Rank函数友好协作（也有可能是我使用的方式不科学呢），而且，在以下呈现的脚本中，我们还生成了一个常量字符串的distribute_key；

2）然后在外层循环中更需要考虑上面3个问题。

请见HQL脚本：

[sql] view
plaincopyprint?





select

device_rank,

device_info,

vv_rank,

pageID,

act_vv

from

(

select

device_rank,

device_info,

(Rank(device_rank) + 1) as vv_rank,

pageID,

act_vv

from

(

select

t2.device_rank,

t2.device_info,

t1.pageID,

t1.act_vv

from

(

select

fieldA as device_info,

pageID,

count(1) as act_vv

from hiveTab_useraction

where `date` >= dateStart and `date` <= dateEnd

group by fieldA, pageID

) t1

join

(

select

(Rank(distribute_key) + 1) as device_rank,

device_info,

act_uv

from

(

select

distribute_key,

device_info,

act_uv

from

(

select

'topdevice' as distribute_key,

device_info,

act_uv

from

(

select

fieldA as device_info,

count(distinct uid) as act_uv

from hiveTab_useraction

where `date` >= dateStart and `date` <= dateEnd

group by fieldA

) t

order by act_uv desc

limit 10

) t

distribute by distribute_key

sort by act_uv desc

) t

) t2 on (t1.device_info = t2.device_info)

distribute by t2.device_rank

sort by t2.device_rank, t1.act_vv desc

) t

) t

where vv_rank <= 10

从脚本实测来看，上面提到的需要明确的3个问题，真的很重要。另外，Rank函数返回的名次是从0开始，所以我们需要作+1处理。

Hive向普通用户也开放了自行编写、注册和使用自定义函数的功能，这一点确实带来了很大的扩展性。