Group By高級用法Groupings Sets語句的功能和底層實現(xiàn)

前言

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

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

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

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

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

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

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

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

Grouping Sets 簡介

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

比如，對于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í)行計劃

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

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

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

首先，我們通過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í)行每個子查詢語句，計算得出查詢結(jié)果。其中，每個查詢語句的邏輯是這樣的：

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

• 在Union節(jié)點完成對每個子查詢結(jié)果的聯(lián)合。
• 最后，在Sort節(jié)點完成對數(shù)據(jù)的排序，上述 Plan 中Sort [city#93 ASC NULLS FIRST, car_model#94 ASC NULLS FIRST]就表示根據(jù)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 版本要簡潔很多！具體的執(zhí)行邏輯是這樣的：

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

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

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

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

通過 Physical Plan，我們發(fā)現(xiàn) Expand 節(jié)點對應(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:[]]

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

Expand 算子的實現(xiàn)

Expand 算子在 Spark SQL 源碼中的實現(xiàn)為ExpandExec類（Spark SQL 中的算子實現(xiàn)類的命名都是XxxExec的格式，其中Xxx為具體的算子名，比如 Project 算子的實現(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)鍵點1，將child.output，也即上游算子輸出數(shù)據(jù)的schema，
//綁定到表達式數(shù)組exprs，以此來計算輸出數(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)鍵點2，projections對應(yīng)了GroupingSets里面每個groupingset的表達式，
//表達式輸出數(shù)據(jù)的schema為this.output,比如(quantity,city,car_model,spark_grouping_id)
//這里的邏輯是為它們各自生成一個UnsafeProjection對象，通過該對象的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)鍵點3，對于輸入數(shù)據(jù)的每一條記錄，都重復(fù)使用N次，其中N的大小對應(yīng)了projections數(shù)組的大小，
//也即GroupingSets里指定的groupingset的數(shù)量
if(idx<=?0){
//intheinitial(-1)orbeginning(0)ofanewinputrow,fetchthenextinputtuple
input=iter.next()
idx=0
}
//關(guān)鍵點4，對輸入數(shù)據(jù)的每一條記錄，通過UnsafeProjection計算得出輸出數(shù)據(jù)，
//每個groupingset對應(yīng)的UnsafeProjection都會對同一個input計算一遍
result=groups(idx)(input)
idx+=1

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

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

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

關(guān)鍵點 1和關(guān)鍵點 2是基礎(chǔ)，關(guān)鍵點 2中的groups是一個UnsafeProjection[N]數(shù)組類型，其中每個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)建時的傳參exprs和inputSchema，對輸入記錄進行列投影，得出輸出記錄。

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

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

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

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

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

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

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

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

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

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

3. spark_grouping_id列的作用是什么？

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

查詢性能對比

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

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

//GroupingSets版本執(zhí)行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版本執(zhí)行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一節(jié)對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的簡寫，等價規(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底層實現(xiàn)用的也是 Expand 算子，說明RollUp確實是基于Grouping Sets實現(xiàn)的。 而且Expand [[quantity#2159, city#2157, car_model#2158, 0], [quantity#2159, city#2157, null, 1], [quantity#2159, null, null, 3]]也表明RollUp符合等價規(guī)則。

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

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

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

最后

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

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

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

文章配圖

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