hive中selectcount(distinctxx)fromtable查询慢原因及优化

一.问题描述

在hive中&＃xff0c;如果数据量很大&＃xff0c;select count(distinct xx) from table 有时会查询非常慢
举例说明&＃xff1a;
表名&＃xff1a;loan_trans
数据量&＃xff1a;551353635
存储空间&＃xff1a;52.93GB

select count(distinct id)
from loan_trans
where etl_tx_dt &＃61;20200202


number of mappers: 228; number of reducers: 1
执行开始时间-结束时间&＃xff1a; 7:48-7:57
执行结果&＃xff1a;3282768

二.分析原因

该语句转化为MapReduce作业后执行示意图如下

由于引入了DISTINCT&＃xff0c;因此在Map阶段无法利用combine对输出结果消重&＃xff0c;必须将id作为Key输出&＃xff0c;在Reduce阶段再对来自于不同Map Task、相同Key的结果进行消重&＃xff0c;计入最终统计值。
我们看到作业运行时的Reduce Task个数为1&＃xff0c;对于统计大数据量时&＃xff0c;这会导致最终Map的全部输出由单个的ReduceTask处理。这唯一的Reduce Task需要Shuffle大量的数据&＃xff0c;并且进行排序聚合等处理&＃xff0c;这使得它成为整个作业的IO和运算瓶颈。
经过上述分析后&＃xff0c;我们尝试显式地增大Reduce Task个数来提高Reduce阶段的并发&＃xff0c;使每一个Reduce Task的数据处理量控制在2G左右。具体设置如下&＃xff1a;

set mapred.reduce.tasks&＃61;100


调整后我们发现这一参数并没有影响实际Reduce Task个数&＃xff0c;Hive运行时输出“Number of reduce tasks determined at compile time: 1”。原来Hive在处理COUNT这种“全聚合(full aggregates)”计算时&＃xff0c;它会忽略用户指定的Reduce Task数&＃xff0c;而强制使用1。

三.优化

我们只能采用变通的方法来绕过这一限制。我们利用Hive对嵌套语句的支持&＃xff0c;将原来一个MapReduce作业转换为两个作业&＃xff0c;**在第一阶段选出全部的非重复id&＃xff0c;在第二阶段再对这些已消重的id进行计数。这样在第一阶段我们可以通过增大Reduce的并发数&＃xff0c;并发处理Map输出。在第二阶段&＃xff0c;由于id已经消重&＃xff0c;因此COUNT(*)操作在Map阶段不需要输出原id数据&＃xff0c;只输出一个合并后的计数即可。这样即使第二阶段Hive强制指定一个Reduce Task&＃xff0c;极少量的Map输出数据也不会使单一的Reduce Task成为瓶颈。**改进后的SQL语句如下&＃xff1a;

select count(*)
from
(
select
id
from loan_trans
where etl_tx_dt &＃61;20200202
group by id
) t


或者

select count(*)
from
(
select
distinct
id
from loan_trans
where etl_tx_dt &＃61;20200202
) t


二者的执行计划一致
执行计划如下&＃xff1a;

STAGE DEPENDENCIES:Stage-1 is a root stageStage-2 depends on stages: Stage-1Stage-0 depends on stages: Stage-2
""
STAGE PLANS:Stage: Stage-1Map ReduceMap Operator Tree:TableScanalias: loan_transfilterExpr: (etl_tx_dt &＃61; 20200202) (type: boolean)Statistics: Num rows: 551353635 Data size: 20400084495 Basic stats: COMPLETE Column stats: NONESelect Operatorexpressions: id (type: bigint)outputColumnNames: idStatistics: Num rows: 551353635 Data size: 20400084495 Basic stats: COMPLETE Column stats: NONEGroup By Operatorkeys: id (type: bigint)mode: hashoutputColumnNames: _col0Statistics: Num rows: 551353635 Data size: 20400084495 Basic stats: COMPLETE Column stats: NONEReduce Output Operatorkey expressions: _col0 (type: bigint)sort order: &＃43;Map-reduce partition columns: _col0 (type: bigint)Statistics: Num rows: 551353635 Data size: 20400084495 Basic stats: COMPLETE Column stats: NONEReduce Operator Tree:Group By Operatorkeys: KEY._col0 (type: bigint)mode: mergepartialoutputColumnNames: _col0Statistics: Num rows: 275676817 Data size: 10200042229 Basic stats: COMPLETE Column stats: NONESelect OperatorStatistics: Num rows: 275676817 Data size: 10200042229 Basic stats: COMPLETE Column stats: NONEGroup By Operatoraggregations: count()mode: hashoutputColumnNames: _col0Statistics: Num rows: 1 Data size: 8 Basic stats: COMPLETE Column stats: NONEFile Output Operatorcompressed: falsetable:input format: org.apache.hadoop.mapred.SequenceFileInputFormatoutput format: org.apache.hadoop.hive.ql.io.HiveSequenceFileOutputFormatserde: org.apache.hadoop.hive.serde2.lazybinary.LazyBinarySerDe
""Stage: Stage-2Map ReduceMap Operator Tree:TableScanReduce Output Operatorsort order: Statistics: Num rows: 1 Data size: 8 Basic stats: COMPLETE Column stats: NONEvalue expressions: _col0 (type: bigint)Reduce Operator Tree:Group By Operatoraggregations: count(VALUE._col0)mode: mergepartialoutputColumnNames: _col0Statistics: Num rows: 1 Data size: 8 Basic stats: COMPLETE Column stats: NONEFile Output Operatorcompressed: falseStatistics: Num rows: 1 Data size: 8 Basic stats: COMPLETE Column stats: NONEtable:input format: org.apache.hadoop.mapred.TextInputFormatoutput format: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormatserde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
""Stage: Stage-0Fetch Operatorlimit: -1Processor Tree:ListSink
""


在实际运行时&＃xff0c;我们发现Hive还对这两阶段的作业做了额外的优化。它将第二个MapReduce作业Map中的Count过程移到了第一个作业的Reduce阶段。这样在第一阶Reduce就可以输出计数值&＃xff0c;而不是消重的全部id。这一优化大幅地减少了第一个作业的Reduce输出IO以及第二个作业Map的输入数据量。最终在同样的运行环境下优化后的语句执行只需要原语句20%左右的时间。优化后的MapReduce作业流如下&＃xff1a;

执行结果&＃xff1a;
执行进度: INFO : Stage-Stage-1: Map: 228 Reduce: 847 Cumulative CPU: 13120.27 sec HDFS Read: 2831413787 HDFS Write: 98252 SUCCESS
执行进度: INFO : Stage-Stage-2: Map: 142 Reduce: 1 Cumulative CPU: 367.8 sec HDFS Read: 544869 HDFS Write: 8 SUCCESS
执行开始时间-结束时间10:12-10:14
结果&＃xff1a;3282768

四.结论

第二次运行结果明显比第一次快了5倍&＃xff0c;所以在需要去重计算时
可以使用 select count(*) from (select distinct id from tablename)
替换
select count(distinct id) from tablename