前言

SQL 中Group By語句大家都很熟悉，根據(jù)指定的規(guī)則對(duì)數(shù)據(jù)進(jìn)行分組，常常和聚合函數(shù)一起使用。

比如，考慮有表dealer，表中數(shù)據(jù)如下：

如果執(zhí)行 SQL 語句SELECT id, sum(quantity) FROM dealer GROUP BY id ORDER BY id，會(huì)得到如下結(jié)果：

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

上述 SQL 語句的意思就是對(duì)數(shù)據(jù)按id列進(jìn)行分組，然后在每個(gè)分組內(nèi)對(duì)quantity列進(jìn)行求和。

Group By語句除了上面的簡(jiǎn)單用法之外，還有更高級(jí)的用法，常見的是Grouping Sets、RollUp和Cube，它們?cè)?OLAP 時(shí)比較常用。其中，RollUp和Cube都是以Grouping Sets為基礎(chǔ)實(shí)現(xiàn)的，因此，弄懂了Grouping Sets，也就理解了RollUp和Cube。

本文首先簡(jiǎn)單介紹Grouping Sets的用法，然后以 Spark SQL 作為切入點(diǎn)，深入解析Grouping Sets的實(shí)現(xiàn)機(jī)制。

Spark SQL 是 Apache Spark 大數(shù)據(jù)處理框架的一個(gè)子模塊，用來處理結(jié)構(gòu)化信息。它可以將 SQL 語句翻譯多個(gè)任務(wù)在 Spark 集群上執(zhí)行，允許用戶直接通過 SQL 來處理數(shù)據(jù)，大大提升了易用性。

Grouping Sets 簡(jiǎn)介

Spark SQL 官方文檔中SQL Syntax一節(jié)對(duì)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語句的作用是指定幾個(gè)grouping set作為Group By的分組規(guī)則，然后再將結(jié)果聯(lián)合在一起。它的效果和，先分別對(duì)這些 grouping set 進(jìn)行Group By分組之后，再通過 Union All 將結(jié)果聯(lián)合起來，是一樣的。

比如，對(duì)于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 版的執(zhí)行結(jié)果：

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 版的執(zhí)行結(jié)果：

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|
+--------+------------+---+

兩版的查詢結(jié)果完全一樣。

Grouping Sets 的執(zhí)行計(jì)劃

從執(zhí)行結(jié)果上看，Grouping Sets 版本和 Union All 版本的 SQL 是等價(jià)的，但 Grouping Sets 版本更加簡(jiǎn)潔。

那么，Grouping Sets僅僅只是Union All的一個(gè)縮寫，或者語法糖嗎？

為了進(jìn)一步探究Grouping Sets的底層實(shí)現(xiàn)是否和Union All是一致的，我們可以來看下兩者的執(zhí)行計(jì)劃。

首先，我們通過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 版本的執(zhí)行邏輯：

1. 執(zhí)行每個(gè)子查詢語句，計(jì)算得出查詢結(jié)果。其中，每個(gè)查詢語句的邏輯是這樣的：

• 在HiveTableRelation節(jié)點(diǎn)對(duì)dealer表進(jìn)行全表掃描。
• 在Project節(jié)點(diǎn)選出與查詢語句結(jié)果相關(guān)的列，比如對(duì)于子查詢語句SELECT NULL as city, NULL as car_model, sum(quantity) AS sum FROM dealer，只需保留quantity列即可。
• 在Aggregate節(jié)點(diǎn)完成quantity列對(duì)聚合運(yùn)算。在上述的 Plan 中，Aggregate 后面緊跟的就是用來分組的列，比如Aggregate [city#902]就表示根據(jù)city列來進(jìn)行分組。

• 在Union節(jié)點(diǎn)完成對(duì)每個(gè)子查詢結(jié)果的聯(lián)合。
• 最后，在Sort節(jié)點(diǎn)完成對(duì)數(shù)據(jù)的排序，上述 Plan 中Sort [city#93 ASC NULLS FIRST, car_model#94 ASC NULLS FIRST]就表示根據(jù)city和car_model列進(jìn)行升序排序。

接下來，我們通過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 版本要簡(jiǎn)潔很多！具體的執(zhí)行邏輯是這樣的：

1. 在HiveTableRelation節(jié)點(diǎn)對(duì)dealer表進(jìn)行全表掃描。
2. 在Project節(jié)點(diǎn)選出與查詢語句結(jié)果相關(guān)的列。
3. 接下來的Expand節(jié)點(diǎn)是關(guān)鍵，數(shù)據(jù)經(jīng)過該節(jié)點(diǎn)后，多出了spark_grouping_id列。從 Plan 中可以看出來，Expand 節(jié)點(diǎn)包含了Grouping Sets里的各個(gè) grouping set 信息，比如[quantity#133, city#131, null, 1]對(duì)應(yīng)的就是(city)這一 grouping set。而且，每個(gè) grouping set 對(duì)應(yīng)的spark_grouping_id列的值都是固定的，比如(city)對(duì)應(yīng)的spark_grouping_id為1。
4. 在Aggregate節(jié)點(diǎn)完成quantity列對(duì)聚合運(yùn)算，其中分組的規(guī)則為city, car_model, spark_grouping_id。注意，數(shù)據(jù)經(jīng)過 Aggregate 節(jié)點(diǎn)后，spark_grouping_id列被刪除了！
5. 最后，在Sort節(jié)點(diǎn)完成對(duì)數(shù)據(jù)的排序。

從 Optimized Logical Plan 來看，雖然 Union All 版本和 Grouping Sets 版本的效果一致，但它們的底層實(shí)現(xiàn)有著巨大的差別。

其中，Grouping Sets 版本的 Plan 中最關(guān)鍵的是 Expand 節(jié)點(diǎn)，目前，我們只知道數(shù)據(jù)經(jīng)過它之后，多出了spark_grouping_id列。而且從最終結(jié)果來看，spark_grouping_id只是 Spark SQL 的內(nèi)部實(shí)現(xiàn)細(xì)節(jié)，對(duì)用戶并不體現(xiàn)。那么：

1. Expand 的實(shí)現(xiàn)邏輯是怎樣的，為什么能達(dá)到Union All的效果？
2. Expand 節(jié)點(diǎn)的輸出數(shù)據(jù)是怎樣的？
3. spark_grouping_id列的作用是什么？

通過 Physical Plan，我們發(fā)現(xiàn) Expand 節(jié)點(diǎn)對(duì)應(yīng)的算子名稱也是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:[]]

帶著前面的幾個(gè)問題，接下來我們深入 Spark SQL 的Expand算子源碼尋找答案。

Expand 算子的實(shí)現(xiàn)

Expand 算子在 Spark SQL 源碼中的實(shí)現(xiàn)為ExpandExec類（Spark SQL 中的算子實(shí)現(xiàn)類的命名都是XxxExec的格式，其中Xxx為具體的算子名，比如 Project 算子的實(shí)現(xiàn)類為ProjectExec），核心代碼如下：

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

...
//關(guān)鍵點(diǎn)1，將child.output，也即上游算子輸出數(shù)據(jù)的schema，
//綁定到表達(dá)式數(shù)組exprs，以此來計(jì)算輸出數(shù)據(jù)
private[this]valprojection=
(exprs:Seq[Expression])=>UnsafeProjection.create(exprs,child.output)

//doExecute()方法為Expand算子執(zhí)行邏輯所在
protectedoverridedefdoExecute():RDD[InternalRow]={
valnumOutputRows=longMetric("numOutputRows")

//處理上游算子的輸出數(shù)據(jù)，Expand算子的輸入數(shù)據(jù)就從iter迭代器獲取
child.execute().mapPartitions{iter=>
//關(guān)鍵點(diǎn)2，projections對(duì)應(yīng)了GroupingSets里面每個(gè)groupingset的表達(dá)式，
//表達(dá)式輸出數(shù)據(jù)的schema為this.output,比如(quantity,city,car_model,spark_grouping_id)
//這里的邏輯是為它們各自生成一個(gè)UnsafeProjection對(duì)象，通過該對(duì)象的apply方法就能得出Expand算子的輸出數(shù)據(jù)
valgroups=projections.map(projection).toArray
newIterator[InternalRow]{
private[this]varresult:InternalRow=_
private[this]varidx=-1//-1meanstheinitialstate
private[this]varinput:InternalRow=_

overridefinaldefhasNext:Boolean=(-1overridefinaldefnext():InternalRow={
//關(guān)鍵點(diǎn)3，對(duì)于輸入數(shù)據(jù)的每一條記錄，都重復(fù)使用N次，其中N的大小對(duì)應(yīng)了projections數(shù)組的大小，
//也即GroupingSets里指定的groupingset的數(shù)量
if(idx<=?0){
//intheinitial(-1)orbeginning(0)ofanewinputrow,fetchthenextinputtuple
input=iter.next()
idx=0
}
//關(guān)鍵點(diǎn)4，對(duì)輸入數(shù)據(jù)的每一條記錄，通過UnsafeProjection計(jì)算得出輸出數(shù)據(jù)，
//每個(gè)groupingset對(duì)應(yīng)的UnsafeProjection都會(huì)對(duì)同一個(gè)input計(jì)算一遍
result=groups(idx)(input)
idx+=1

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

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

ExpandExec的實(shí)現(xiàn)并不復(fù)雜，想要理解它的運(yùn)作原理，關(guān)鍵是看懂上述源碼中提到的 4 個(gè)關(guān)鍵點(diǎn)。

關(guān)鍵點(diǎn) 1和關(guān)鍵點(diǎn) 2是基礎(chǔ)，關(guān)鍵點(diǎn) 2中的groups是一個(gè)UnsafeProjection[N]數(shù)組類型，其中每個(gè)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方法根據(jù)創(chuàng)建時(shí)的傳參exprs和inputSchema，對(duì)輸入記錄進(jìn)行列投影，得出輸出記錄。

比如，前面的GROUPING SETS ((city, car_model), (city), (car_model), ())例子，它對(duì)應(yīng)的groups是這樣的：

其中，AttributeReference類型的表達(dá)式，在計(jì)算時(shí)，會(huì)直接引用輸入數(shù)據(jù)對(duì)應(yīng)列的值；Iteral類型的表達(dá)式，在計(jì)算時(shí)，值是固定的。

關(guān)鍵點(diǎn) 3和關(guān)鍵點(diǎn) 4是 Expand 算子的精華所在，ExpandExec通過這兩段邏輯，將每一個(gè)輸入記錄，擴(kuò)展（Expand）成 N 條輸出記錄。

關(guān)鍵點(diǎn) 4中groups(idx)(input)等同于groups(idx).apply(input)。

還是以前面GROUPING SETS ((city, car_model), (city), (car_model), ())為例子，效果是這樣的：

到這里，我們已經(jīng)弄清楚 Expand 算子的工作原理，再回頭看前面提到的 3 個(gè)問題，也不難回答了：

1. Expand 的實(shí)現(xiàn)邏輯是怎樣的，為什么能達(dá)到Union All的效果？

   如果說Union All是先聚合再聯(lián)合，那么 Expand 就是先聯(lián)合再聚合。Expand 利用groups里的 N 個(gè)表達(dá)式對(duì)每條輸入記錄進(jìn)行計(jì)算，擴(kuò)展成 N 條輸出記錄。后面再聚合時(shí)，就能達(dá)到與Union All一樣的效果了。

2. Expand 節(jié)點(diǎn)的輸出數(shù)據(jù)是怎樣的？

   在 schema 上，Expand 輸出數(shù)據(jù)會(huì)比輸入數(shù)據(jù)多出spark_grouping_id列；在記錄數(shù)上，是輸入數(shù)據(jù)記錄數(shù)的 N 倍。

3. spark_grouping_id列的作用是什么？

   spark_grouping_id給每個(gè) grouping set 進(jìn)行編號(hào)，這樣，即使在 Expand 階段把數(shù)據(jù)先聯(lián)合起來，在 Aggregate 階段（把spark_grouping_id加入到分組規(guī)則）也能保證數(shù)據(jù)能夠按照每個(gè) grouping set 分別聚合，確保了結(jié)果的正確性。

查詢性能對(duì)比

從前文可知，Grouping Sets 和 Union All 兩個(gè)版本的 SQL 語句有著一樣的效果，但是它們的執(zhí)行計(jì)劃卻有著巨大的差別。下面，我們將比對(duì)兩個(gè)版本之間的執(zhí)行性能差異。

spark-sql 執(zhí)行完 SQL 語句之后會(huì)打印耗時(shí)信息，我們對(duì)兩個(gè)版本的 SQL 分別執(zhí)行 10 次，得到如下信息：

//GroupingSets版本執(zhí)行10次的耗時(shí)信息
//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版本執(zhí)行10次的耗時(shí)信息
//(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 平均耗時(shí)為0.276s；Union All 版本的 SQL 平均耗時(shí)為0.616s，是前者的2.2 倍！

所以，Grouping Sets 版本的 SQL 不僅在表達(dá)上更加簡(jiǎn)潔，在性能上也更加高效。

RollUp 和 Cube

Group By的高級(jí)用法中，還有RollUp和Cube兩個(gè)比較常用。

首先，我們看下RollUp語句。

Spark SQL 官方文檔中SQL Syntax一節(jié)對(duì)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的簡(jiǎn)寫，等價(jià)規(guī)則為：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底層實(shí)現(xiàn)用的也是 Expand 算子，說明RollUp確實(shí)是基于Grouping Sets實(shí)現(xiàn)的。 而且Expand [[quantity#2159, city#2157, car_model#2158, 0], [quantity#2159, city#2157, null, 1], [quantity#2159, null, null, 3]]也表明RollUp符合等價(jià)規(guī)則。

下面，我們按照同樣的思路，看下Cube語句。

Spark SQL 官方文檔中SQL Syntax一節(jié)對(duì)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的簡(jiǎn)寫，等價(jià)規(guī)則為：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確實(shí)基于Grouping Sets實(shí)現(xiàn)，而且也符合等價(jià)規(guī)則。

所以，RollUp和Cube可以看成是Grouping Sets的語法糖，在底層實(shí)現(xiàn)和性能上是一樣的。

最后

本文重點(diǎn)討論了Group By高級(jí)用法Groupings Sets語句的功能和底層實(shí)現(xiàn)。

雖然Groupings Sets的功能，通過Union All也能實(shí)現(xiàn)，但前者并非后者的語法糖，它們的底層實(shí)現(xiàn)完全不一樣。Grouping Sets采用的是先聯(lián)合再聚合的思路，通過spark_grouping_id列來保證數(shù)據(jù)的正確性；Union All則采用先聚合再聯(lián)合的思路。Grouping Sets在 SQL 語句表達(dá)和性能上都有更大的優(yōu)勢(shì)。

Group By的另外兩個(gè)高級(jí)用法RollUp和Cube則可以看成是Grouping Sets的語法糖，它們的底層都是基于 Expand 算子實(shí)現(xiàn)，在性能上與直接使用Grouping Sets是一樣的，但在 SQL 表達(dá)上更加簡(jiǎn)潔。

文章配圖

可以在用Keynote畫出手繪風(fēng)格的配圖中找到文章的繪圖方法。