Group By高级用法Groupings Sets语句的功能和底层实现-电子发烧友网

前言

SQL 中Group By语句大家都很熟悉，根据指定的规则对数据进行分组，常常和聚合函数一起使用。

比如，考虑有表dealer，表中数据如下：

id (Int)	city (String)	car_model (String)	quantity (Int)
100	Fremont	Honda Civic	10
100	Fremont	Honda Accord	15
100	Fremont	Honda CRV	7
200	Dublin	Honda Civic	20
200	Dublin	Honda Accord	10
200	Dublin	Honda CRV	3
300	San Jose	Honda Civic	5
300	San Jose	Honda Accord	8

如果执行 SQL 语句SELECT id, sum(quantity) FROM dealer GROUP BY id ORDER BY id，会得到如下结果：

+---+-------------+
|id|sum(quantity)|
+---+-------------+
|100|32|
|200|33|
|300|13|
+---+-------------+

上述 SQL 语句的意思就是对数据按id列进行分组，然后在每个分组内对quantity列进行求和。

Group By语句除了上面的简单用法之外，还有更高级的用法，常见的是Grouping Sets、RollUp和Cube，它们在 OLAP 时比较常用。其中，RollUp和Cube都是以Grouping Sets为基础实现的，因此，弄懂了Grouping Sets，也就理解了RollUp和Cube。

本文首先简单介绍Grouping Sets的用法，然后以 Spark SQL 作为切入点，深入解析Grouping Sets的实现机制。

Spark SQL 是 Apache Spark 大数据处理框架的一个子模块，用来处理结构化信息。它可以将 SQL 语句翻译多个任务在 Spark 集群上执行，允许用户直接通过 SQL 来处理数据，大大提升了易用性。

Grouping Sets 简介

Spark SQL 官方文档中SQL Syntax一节对Grouping Sets语句的描述如下：

Groups the rows for each grouping set specified after GROUPING SETS. （... 一些举例） This clause is a shorthand for aUNION ALLwhere each leg of theUNION ALLoperator performs aggregation of each grouping set specified in theGROUPING SETSclause. （... 一些举例）

也即，Grouping Sets语句的作用是指定几个grouping set作为Group By的分组规则，然后再将结果联合在一起。它的效果和，先分别对这些 grouping set 进行Group By分组之后，再通过 Union All 将结果联合起来，是一样的。

比如，对于dealer表，Group By Grouping Sets ((city, car_model), (city), (car_model), ())和Union All((Group By city, car_model), (Group By city), (Group By car_model), 全局聚合)的效果是相同的：

先看 Grouping Sets 版的执行结果：

spark-sql>SELECTcity,car_model,sum(quantity)ASsumFROMdealer
>GROUPBYGROUPINGSETS((city,car_model),(city),(car_model),())
>ORDERBYcity,car_model;
+--------+------------+---+
|city|car_model|sum|
+--------+------------+---+
|null|null|78|
|null|HondaAccord|33|
|null|HondaCRV|10|
|null|HondaCivic|35|
|Dublin|null|33|
|Dublin|HondaAccord|10|
|Dublin|HondaCRV|3|
|Dublin|HondaCivic|20|
|Fremont|null|32|
|Fremont|HondaAccord|15|
|Fremont|HondaCRV|7|
|Fremont|HondaCivic|10|
|SanJose|null|13|
|SanJose|HondaAccord|8|
|SanJose|HondaCivic|5|
+--------+------------+---+

再看 Union All 版的执行结果：

spark-sql>(SELECTcity,car_model,sum(quantity)ASsumFROMdealerGROUPBYcity,car_model)UNIONALL
>(SELECTcity,NULLascar_model,sum(quantity)ASsumFROMdealerGROUPBYcity)UNIONALL
>(SELECTNULLascity,car_model,sum(quantity)ASsumFROMdealerGROUPBYcar_model)UNIONALL
>(SELECTNULLascity,NULLascar_model,sum(quantity)ASsumFROMdealer)
>ORDERBYcity,car_model;
+--------+------------+---+
|city|car_model|sum|
+--------+------------+---+
|null|null|78|
|null|HondaAccord|33|
|null|HondaCRV|10|
|null|HondaCivic|35|
|Dublin|null|33|
|Dublin|HondaAccord|10|
|Dublin|HondaCRV|3|
|Dublin|HondaCivic|20|
|Fremont|null|32|
|Fremont|HondaAccord|15|
|Fremont|HondaCRV|7|
|Fremont|HondaCivic|10|
|SanJose|null|13|
|SanJose|HondaAccord|8|
|SanJose|HondaCivic|5|
+--------+------------+---+

两版的查询结果完全一样。

Grouping Sets 的执行计划

从执行结果上看，Grouping Sets 版本和 Union All 版本的 SQL 是等价的，但 Grouping Sets 版本更加简洁。

那么，Grouping Sets仅仅只是Union All的一个缩写，或者语法糖吗？

为了进一步探究Grouping Sets的底层实现是否和Union All是一致的，我们可以来看下两者的执行计划。

首先，我们通过explain extended来查看 Union All 版本的Optimized Logical Plan:

spark-sql>explainextended(SELECTcity,car_model,sum(quantity)ASsumFROMdealerGROUPBYcity,car_model)UNIONALL(SELECTcity,NULLascar_model,sum(quantity)ASsumFROMdealerGROUPBYcity)UNIONALL(SELECTNULLascity,car_model,sum(quantity)ASsumFROMdealerGROUPBYcar_model)UNIONALL(SELECTNULLascity,NULLascar_model,sum(quantity)ASsumFROMdealer)ORDERBYcity,car_model;
==ParsedLogicalPlan==
...
==AnalyzedLogicalPlan==
...
==OptimizedLogicalPlan==
Sort[city#93ASCNULLSFIRST,car_model#94ASCNULLSFIRST],true
+-Unionfalse,false
:-Aggregate[city#93,car_model#94],[city#93,car_model#94,sum(quantity#95)ASsum#79L]
:+-Project[city#93,car_model#94,quantity#95]
:+-HiveTableRelation[`default`.`dealer`,...,DataCols:[id#92,city#93,car_model#94,quantity#95],PartitionCols:[]]
:-Aggregate[city#97],[city#97,nullAScar_model#112,sum(quantity#99)ASsum#81L]
:+-Project[city#97,quantity#99]
:+-HiveTableRelation[`default`.`dealer`,...,DataCols:[id#96,city#97,car_model#98,quantity#99],PartitionCols:[]]
:-Aggregate[car_model#102],[nullAScity#113,car_model#102,sum(quantity#103)ASsum#83L]
:+-Project[car_model#102,quantity#103]
:+-HiveTableRelation[`default`.`dealer`,...,DataCols:[id#100,city#101,car_model#102,quantity#103],PartitionCols:[]]
+-Aggregate[nullAScity#114,nullAScar_model#115,sum(quantity#107)ASsum#86L]
+-Project[quantity#107]
+-HiveTableRelation[`default`.`dealer`,...,DataCols:[id#104,city#105,car_model#106,quantity#107],PartitionCols:[]]
==PhysicalPlan==
...

从上述的 Optimized Logical Plan 可以清晰地看出 Union All 版本的执行逻辑：

执行每个子查询语句，计算得出查询结果。其中，每个查询语句的逻辑是这样的：

在HiveTableRelation节点对dealer表进行全表扫描。
在Project节点选出与查询语句结果相关的列，比如对于子查询语句SELECT NULL as city, NULL as car_model, sum(quantity) AS sum FROM dealer，只需保留quantity列即可。
在Aggregate节点完成quantity列对聚合运算。在上述的 Plan 中，Aggregate 后面紧跟的就是用来分组的列，比如Aggregate [city#902]就表示根据city列来进行分组。

在Union节点完成对每个子查询结果的联合。
最后，在Sort节点完成对数据的排序，上述 Plan 中Sort [city#93 ASC NULLS FIRST, car_model#94 ASC NULLS FIRST]就表示根据city和car_model列进行升序排序。

接下来，我们通过explain extended来查看 Grouping Sets 版本的 Optimized Logical Plan：

spark-sql>explainextendedSELECTcity,car_model,sum(quantity)ASsumFROMdealerGROUPBYGROUPINGSETS((city,car_model),(city),(car_model),())ORDERBYcity,car_model;
==ParsedLogicalPlan==
...
==AnalyzedLogicalPlan==
...
==OptimizedLogicalPlan==
Sort[city#138ASCNULLSFIRST,car_model#139ASCNULLSFIRST],true
+-Aggregate[city#138,car_model#139,spark_grouping_id#137L],[city#138,car_model#139,sum(quantity#133)ASsum#124L]
+-Expand[[quantity#133,city#131,car_model#132,0],[quantity#133,city#131,null,1],[quantity#133,null,car_model#132,2],[quantity#133,null,null,3]],[quantity#133,city#138,car_model#139,spark_grouping_id#137L]
+-Project[quantity#133,city#131,car_model#132]
+-HiveTableRelation[`default`.`dealer`,org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe,DataCols:[id#130,city#131,car_model#132,quantity#133],PartitionCols:[]]
==PhysicalPlan==
...

从 Optimized Logical Plan 来看，Grouping Sets 版本要简洁很多！具体的执行逻辑是这样的：

在HiveTableRelation节点对dealer表进行全表扫描。
在Project节点选出与查询语句结果相关的列。
接下来的Expand节点是关键，数据经过该节点后，多出了spark_grouping_id列。从 Plan 中可以看出来，Expand 节点包含了Grouping Sets里的各个 grouping set 信息，比如[quantity#133, city#131, null, 1]对应的就是(city)这一 grouping set。而且，每个 grouping set 对应的spark_grouping_id列的值都是固定的，比如(city)对应的spark_grouping_id为1。
在Aggregate节点完成quantity列对聚合运算，其中分组的规则为city, car_model, spark_grouping_id。注意，数据经过 Aggregate 节点后，spark_grouping_id列被删除了！
最后，在Sort节点完成对数据的排序。

从 Optimized Logical Plan 来看，虽然 Union All 版本和 Grouping Sets 版本的效果一致，但它们的底层实现有着巨大的差别。

其中，Grouping Sets 版本的 Plan 中最关键的是 Expand 节点，目前，我们只知道数据经过它之后，多出了spark_grouping_id列。而且从最终结果来看，spark_grouping_id只是 Spark SQL 的内部实现细节，对用户并不体现。那么：

Expand 的实现逻辑是怎样的，为什么能达到Union All的效果？
Expand 节点的输出数据是怎样的？
spark_grouping_id列的作用是什么？

通过 Physical Plan，我们发现 Expand 节点对应的算子名称也是Expand:

==PhysicalPlan==
AdaptiveSparkPlanisFinalPlan=false
+-Sort[city#138ASCNULLSFIRST,car_model#139ASCNULLSFIRST],true,0
+-Exchangerangepartitioning(city#138ASCNULLSFIRST,car_model#139ASCNULLSFIRST,200),ENSURE_REQUIREMENTS,[plan_id=422]
+-HashAggregate(keys=[city#138,car_model#139,spark_grouping_id#137L],functions=[sum(quantity#133)],output=[city#138,car_model#139,sum#124L])
+-Exchangehashpartitioning(city#138,car_model#139,spark_grouping_id#137L,200),ENSURE_REQUIREMENTS,[plan_id=419]
+-HashAggregate(keys=[city#138,car_model#139,spark_grouping_id#137L],functions=[partial_sum(quantity#133)],output=[city#138,car_model#139,spark_grouping_id#137L,sum#141L])
+-Expand[[quantity#133,city#131,car_model#132,0],[quantity#133,city#131,null,1],[quantity#133,null,car_model#132,2],[quantity#133,null,null,3]],[quantity#133,city#138,car_model#139,spark_grouping_id#137L]
+-Scanhivedefault.dealer[quantity#133,city#131,car_model#132],HiveTableRelation[`default`.`dealer`,...,DataCols:[id#130,city#131,car_model#132,quantity#133],PartitionCols:[]]

带着前面的几个问题，接下来我们深入 Spark SQL 的Expand算子源码寻找答案。

Expand 算子的实现

Expand 算子在 Spark SQL 源码中的实现为ExpandExec类（Spark SQL 中的算子实现类的命名都是XxxExec的格式，其中Xxx为具体的算子名，比如 Project 算子的实现类为ProjectExec），核心代码如下：

/**
*ApplyalloftheGroupExpressionstoeveryinputrow,hencewewillget
*multipleoutputrowsforaninputrow.
*@paramprojectionsThegroupofexpressions,allofthegroupexpressionsshould
*outputthesameschemaspecifiedbyetheparameter`output`
*@paramoutputTheoutputSchema
*@paramchildChildoperator
*/
caseclassExpandExec(
projections:Seq[Seq[Expression]],
output:Seq[Attribute],
child:SparkPlan)
extendsUnaryExecNodewithCodegenSupport{

...
//关键点1，将child.output，也即上游算子输出数据的schema，
//绑定到表达式数组exprs，以此来计算输出数据
private[this]valprojection=
(exprs:Seq[Expression])=>UnsafeProjection.create(exprs,child.output)

//doExecute()方法为Expand算子执行逻辑所在
protectedoverridedefdoExecute():RDD[InternalRow]={
valnumOutputRows=longMetric("numOutputRows")

//处理上游算子的输出数据，Expand算子的输入数据就从iter迭代器获取
child.execute().mapPartitions{iter=>
//关键点2，projections对应了GroupingSets里面每个groupingset的表达式，
//表达式输出数据的schema为this.output,比如(quantity,city,car_model,spark_grouping_id)
//这里的逻辑是为它们各自生成一个UnsafeProjection对象，通过该对象的apply方法就能得出Expand算子的输出数据
valgroups=projections.map(projection).toArray
newIterator[InternalRow]{
private[this]varresult:InternalRow=_
private[this]varidx=-1//-1meanstheinitialstate
private[this]varinput:InternalRow=_

overridefinaldefhasNext:Boolean=(-1< idx && idx < groups.length) || iter.hasNext

        overridefinaldefnext():InternalRow={
//关键点3，对于输入数据的每一条记录，都重复使用N次，其中N的大小对应了projections数组的大小，
//也即GroupingSets里指定的groupingset的数量
if(idx<= 0){
//intheinitial(-1)orbeginning(0)ofanewinputrow,fetchthenextinputtuple
input=iter.next()
idx=0
}
//关键点4，对输入数据的每一条记录，通过UnsafeProjection计算得出输出数据，
//每个groupingset对应的UnsafeProjection都会对同一个input计算一遍
result=groups(idx)(input)
idx+=1

if(idx==groups.length&&iter.hasNext){
idx=0
}

numOutputRows+=1
result
}
}
}
}
...
}

ExpandExec的实现并不复杂，想要理解它的运作原理，关键是看懂上述源码中提到的 4 个关键点。

关键点 1和关键点 2是基础，关键点 2中的groups是一个UnsafeProjection[N]数组类型，其中每个UnsafeProjection代表了Grouping Sets语句里指定的 grouping set，它的定义是这样的：

//AprojectionthatreturnsUnsafeRow.
abstractclassUnsafeProjectionextendsProjection{
overridedefapply(row:InternalRow):UnsafeRow
}

//Thefactoryobjectfor`UnsafeProjection`.
objectUnsafeProjection
extendsCodeGeneratorWithInterpretedFallback[Seq[Expression],UnsafeProjection]{
//ReturnsanUnsafeProjectionforgivensequenceofExpressions,whichwillbeboundto
//`inputSchema`.
defcreate(exprs:Seq[Expression],inputSchema:Seq[Attribute]):UnsafeProjection={
create(bindReferences(exprs,inputSchema))
}
...
}

UnsafeProjection起来了类似列投影的作用，其中，apply方法根据创建时的传参exprs和inputSchema，对输入记录进行列投影，得出输出记录。

比如，前面的GROUPING SETS ((city, car_model), (city), (car_model), ())例子，它对应的groups是这样的：

其中，AttributeReference类型的表达式，在计算时，会直接引用输入数据对应列的值；Iteral类型的表达式，在计算时，值是固定的。

关键点 3和关键点 4是 Expand 算子的精华所在，ExpandExec通过这两段逻辑，将每一个输入记录，扩展（Expand）成 N 条输出记录。

关键点 4中groups(idx)(input)等同于groups(idx).apply(input)。

还是以前面GROUPING SETS ((city, car_model), (city), (car_model), ())为例子，效果是这样的：

到这里，我们已经弄清楚 Expand 算子的工作原理，再回头看前面提到的 3 个问题，也不难回答了：

Expand 的实现逻辑是怎样的，为什么能达到Union All的效果？

如果说Union All是先聚合再联合，那么 Expand 就是先联合再聚合。Expand 利用groups里的 N 个表达式对每条输入记录进行计算，扩展成 N 条输出记录。后面再聚合时，就能达到与Union All一样的效果了。
Expand 节点的输出数据是怎样的？

在 schema 上，Expand 输出数据会比输入数据多出spark_grouping_id列；在记录数上，是输入数据记录数的 N 倍。
spark_grouping_id列的作用是什么？

spark_grouping_id给每个 grouping set 进行编号，这样，即使在 Expand 阶段把数据先联合起来，在 Aggregate 阶段（把spark_grouping_id加入到分组规则）也能保证数据能够按照每个 grouping set 分别聚合，确保了结果的正确性。

查询性能对比

从前文可知，Grouping Sets 和 Union All 两个版本的 SQL 语句有着一样的效果，但是它们的执行计划却有着巨大的差别。下面，我们将比对两个版本之间的执行性能差异。

spark-sql 执行完 SQL 语句之后会打印耗时信息，我们对两个版本的 SQL 分别执行 10 次，得到如下信息：

//GroupingSets版本执行10次的耗时信息
//SELECTcity,car_model,sum(quantity)ASsumFROMdealerGROUPBYGROUPINGSETS((city,car_model),(city),(car_model),())ORDERBYcity,car_model;
Timetaken:0.289seconds,Fetched15row(s)
Timetaken:0.251seconds,Fetched15row(s)
Timetaken:0.259seconds,Fetched15row(s)
Timetaken:0.258seconds,Fetched15row(s)
Timetaken:0.296seconds,Fetched15row(s)
Timetaken:0.247seconds,Fetched15row(s)
Timetaken:0.298seconds,Fetched15row(s)
Timetaken:0.286seconds,Fetched15row(s)
Timetaken:0.292seconds,Fetched15row(s)
Timetaken:0.282seconds,Fetched15row(s)

//UnionAll版本执行10次的耗时信息
//(SELECTcity,car_model,sum(quantity)ASsumFROMdealerGROUPBYcity,car_model)UNIONALL(SELECTcity,NULLascar_model,sum(quantity)ASsumFROMdealerGROUPBYcity)UNIONALL(SELECTNULLascity,car_model,sum(quantity)ASsumFROMdealerGROUPBYcar_model)UNIONALL(SELECTNULLascity,NULLascar_model,sum(quantity)ASsumFROMdealer)ORDERBYcity,car_model;
Timetaken:0.628seconds,Fetched15row(s)
Timetaken:0.594seconds,Fetched15row(s)
Timetaken:0.591seconds,Fetched15row(s)
Timetaken:0.607seconds,Fetched15row(s)
Timetaken:0.616seconds,Fetched15row(s)
Timetaken:0.64seconds,Fetched15row(s)
Timetaken:0.623seconds,Fetched15row(s)
Timetaken:0.625seconds,Fetched15row(s)
Timetaken:0.62seconds,Fetched15row(s)
Timetaken:0.62seconds,Fetched15row(s)

可以算出，Grouping Sets 版本的 SQL 平均耗时为0.276s；Union All 版本的 SQL 平均耗时为0.616s，是前者的2.2 倍！

所以，Grouping Sets 版本的 SQL 不仅在表达上更加简洁，在性能上也更加高效。

RollUp 和 Cube

Group By的高级用法中，还有RollUp和Cube两个比较常用。

首先，我们看下RollUp语句。

Spark SQL 官方文档中SQL Syntax一节对RollUp语句的描述如下：

Specifies multiple levels of aggregations in a single statement. This clause is used to compute aggregations based on multiple grouping sets.ROLLUPis a shorthand forGROUPING SETS. （... 一些例子）

官方文档中，把RollUp描述为Grouping Sets的简写，等价规则为：RollUp(A, B, C) == Grouping Sets((A, B, C), (A, B), (A), ())。

比如，Group By RollUp(city, car_model)就等同于Group By Grouping Sets((city, car_model), (city), ())。

下面，我们通过expand extended看下 RollUp 版本 SQL 的 Optimized Logical Plan：

spark-sql>explainextendedSELECTcity,car_model,sum(quantity)ASsumFROMdealerGROUPBYROLLUP(city,car_model)ORDERBYcity,car_model;
==ParsedLogicalPlan==
...
==AnalyzedLogicalPlan==
...
==OptimizedLogicalPlan==
Sort[city#2164ASCNULLSFIRST,car_model#2165ASCNULLSFIRST],true
+-Aggregate[city#2164,car_model#2165,spark_grouping_id#2163L],[city#2164,car_model#2165,sum(quantity#2159)ASsum#2150L]
+-Expand[[quantity#2159,city#2157,car_model#2158,0],[quantity#2159,city#2157,null,1],[quantity#2159,null,null,3]],[quantity#2159,city#2164,car_model#2165,spark_grouping_id#2163L]
+-Project[quantity#2159,city#2157,car_model#2158]
+-HiveTableRelation[`default`.`dealer`,...,DataCols:[id#2156,city#2157,car_model#2158,quantity#2159],PartitionCols:[]]
==PhysicalPlan==
...

从上述 Plan 可以看出，RollUp底层实现用的也是 Expand 算子，说明RollUp确实是基于Grouping Sets实现的。而且Expand [[quantity#2159, city#2157, car_model#2158, 0], [quantity#2159, city#2157, null, 1], [quantity#2159, null, null, 3]]也表明RollUp符合等价规则。

下面，我们按照同样的思路，看下Cube语句。

Spark SQL 官方文档中SQL Syntax一节对Cube语句的描述如下：

CUBEclause is used to perform aggregations based on combination of grouping columns specified in theGROUP BYclause.CUBEis a shorthand forGROUPING SETS. (... 一些例子)

同样，官方文档把Cube描述为Grouping Sets的简写，等价规则为：Cube(A, B, C) == Grouping Sets((A, B, C), (A, B), (A, C), (B, C), (A), (B), (C), ())。

比如，Group By Cube(city, car_model)就等同于Group By Grouping Sets((city, car_model), (city), (car_model), ())。

下面，我们通过expand extended看下 Cube 版本 SQL 的 Optimized Logical Plan：

spark-sql>explainextendedSELECTcity,car_model,sum(quantity)ASsumFROMdealerGROUPBYCUBE(city,car_model)ORDERBYcity,car_model;
==ParsedLogicalPlan==
...
==AnalyzedLogicalPlan==
...
==OptimizedLogicalPlan==
Sort[city#2202ASCNULLSFIRST,car_model#2203ASCNULLSFIRST],true
+-Aggregate[city#2202,car_model#2203,spark_grouping_id#2201L],[city#2202,car_model#2203,sum(quantity#2197)ASsum#2188L]
+-Expand[[quantity#2197,city#2195,car_model#2196,0],[quantity#2197,city#2195,null,1],[quantity#2197,null,car_model#2196,2],[quantity#2197,null,null,3]],[quantity#2197,city#2202,car_model#2203,spark_grouping_id#2201L]
+-Project[quantity#2197,city#2195,car_model#2196]
+-HiveTableRelation[`default`.`dealer`,...,DataCols:[id#2194,city#2195,car_model#2196,quantity#2197],PartitionCols:[]]
==PhysicalPlan==
...

从上述 Plan 可以看出，Cube底层用的也是 Expand 算子，说明Cube确实基于Grouping Sets实现，而且也符合等价规则。

所以，RollUp和Cube可以看成是Grouping Sets的语法糖，在底层实现和性能上是一样的。

最后

本文重点讨论了Group By高级用法Groupings Sets语句的功能和底层实现。

虽然Groupings Sets的功能，通过Union All也能实现，但前者并非后者的语法糖，它们的底层实现完全不一样。Grouping Sets采用的是先联合再聚合的思路，通过spark_grouping_id列来保证数据的正确性；Union All则采用先聚合再联合的思路。Grouping Sets在 SQL 语句表达和性能上都有更大的优势。

Group By的另外两个高级用法RollUp和Cube则可以看成是Grouping Sets的语法糖，它们的底层都是基于 Expand 算子实现，在性能上与直接使用Grouping Sets是一样的，但在 SQL 表达上更加简洁。