[大数据] SPARK SQL中 Grouping sets转Expand怎么实现的（逻辑计划级别）

开发: C++知识库 Java知识库 JavaScript Python PHP知识库人工智能区块链大数据移动开发嵌入式开发工具数据结构与算法开发测试游戏开发网络协议系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑笔记本显卡显示器固态硬盘硬盘耳机手机 iphone vivo oppo 小米华为单反装机图拉丁

-> 大数据 -> SPARK SQL中 Grouping sets转Expand怎么实现的（逻辑计划级别） -> 正文阅读

[大数据]SPARK SQL中 Grouping sets转Expand怎么实现的（逻辑计划级别）

背景

本文基于spark 3.1.2
之前在做bug调试的时候遇到了expand的问题，在此记录一下

分析

运行该sql：

create table test_a_pt(col1 int, col2 int,pt string) USING parquet PARTITIONED BY (pt);
insert into table test_a_pt values(1,2,'20220101'),(3,4,'20220101'),(1,2,'20220101'),(3,4,'20220101'),(1,2,'20220101'),(3,4,'20220101');
select count(*),col1 as alias
from test_a_pt
group by col1,col2
grouping sets (col1,col2)
order by col1,col2 ;

可以看到如下逻辑计划的变化（只截取grouping sets相关的）：

=== Applying Rule org.apache.spark.sql.catalyst.analysis.Analyzer$ResolveRelations ===
 'Sort ['col1 ASC NULLS FIRST], true                                                                          'Sort ['col1 ASC NULLS FIRST], true
 +- 'GroupingSets [ArrayBuffer('col1), ArrayBuffer('col2)], ['col1, 'col2], ['col1, 'count(1) AS alias#221]   +- 'GroupingSets [ArrayBuffer('col1), ArrayBuffer('col2)], ['col1, 'col2], ['col1, 'count(1) AS alias#221]
!   +- 'UnresolvedRelation [test_table], [], false                                                               +- 'SubqueryAlias spark_catalog.default.test_table
!                                                                                                                   +- 'UnresolvedCatalogRelation `default`.`test_table`, [], false

对于GroupingSets里面的信息做一下解释：

'GroupingSets [ArrayBuffer('col1), ArrayBuffer('col2)], ['col1, 'col2], ['col1, 'count(1) AS alias#221]

*`*表示还未解析的计划，
[ArrayBuffer('col1), ArrayBuffer('col2)] 是grouping sets里面的两个值col1和col2
['col1, 'col2] 是group by后面的值col1和col2
['col1, 'count(1) AS alias#221] 是聚合表达式的值，也就是select后面的值 count(*),col1 as alias

接下来就是：
ResolveGroupingAnalytics计划：

06:49:07.323 WARN org.apache.spark.sql.catalyst.rules.PlanChangeLogger: 
=== Applying Rule org.apache.spark.sql.catalyst.analysis.Analyzer$ResolveGroupingAnalytics ===
 'Sort ['col1 ASC NULLS FIRST], true                                                                                         'Sort ['col1 ASC NULLS FIRST], true
!+- 'GroupingSets [ArrayBuffer(col1#223), ArrayBuffer(col2#224)], [col1#223, col2#224], [col1#223, count(1) AS alias#221L]   +- Aggregate [col1#229, col2#230, spark_grouping_id#228L], [col1#229, count(1) AS alias#221L]
!   +- SubqueryAlias spark_catalog.default.test_table                                                                           +- Expand [List(col1#223, col2#224, pt#225, col1#226, null, 1), List(col1#223, col2#224, pt#225, null, col2#227, 2)], [col1#223, col2#224, pt#225, col1#229, col2#230, spark_grouping_id#228L]
!      +- Relation[col1#223,col2#224,pt#225] parquet                                                                               +- Project [col1#223, col2#224, pt#225, col1#223 AS col1#226, col2#224 AS col2#227]
!                                                                                                                                     +- SubqueryAlias spark_catalog.default.test_table
!                                                                                                                                        +- Relation[col1#223,col2#224,pt#225] parquet

代码自己可以去看，我们从逻辑来上分析：

'GroupingSets [ArrayBuffer(col1#223), ArrayBuffer(col2#224)], [col1#223, col2#224], [col1#223, count(1) AS alias#221L] 
                                        ||
                                        \/
 +- Aggregate [col1#229, col2#230, spark_grouping_id#228L], [col1#229, count(1) AS alias#221L]                                   
  +- Expand [List(col1#223, col2#224, pt#225, col1#226, null, 1), List(col1#223, col2#224, pt#225, null, col2#227, 2)], [col1#223, col2#224, pt#225, col1#229, col2#230, spark_grouping_id#228L]
   +- Project [col1#223, col2#224, pt#225, col1#223 AS col1#226, col2#224 AS col2#227]

把最重要的转换提取出来做解释：

+- Project [col1#223, col2#224, pt#225, col1#223 AS col1#226, col2#224 AS col2#227]

前三个expression col1#223, col2#224, pt#225 是根据 Relation(也就是从表test_a_pt直接获取到的，和表的字段保持一致)
后面的expression col1#223 AS col1#226, col2#224 AS col2#227 是根据grouping sets和group by的值整合过来的(并且会加上别名，取别名是为了Expand用的)，如果没有group by 这个表达式才会取grouping sets的值，否则就取group by后面的值(目前spark 3.1.2的做法是group by的属性肯定包含了grouping sets里面的属性,SPARK-33229可以支持)：

如：group by col1,col2 grouping sets (col1,col2) 
则取 col1,col2 

如：grouping sets (col1,col2) 
则取 col1,col2

对于Expand:

Expand [List(col1#223, col2#224, pt#225, col1#226, null, 1), List(col1#223, col2#224, pt#225, null, col2#227, 2)], [col1#223, col2#224, pt#225, col1#229, col2#230, spark_grouping_id#228L]

List(col1#223, col2#224, pt#225, col1#226, null, 1), List(col1#223, col2#224, pt#225, null, col2#227, 2) 这些是expand的输入expression，其中

List(col1#223, col2#224, pt#225, col1#226, null, 1) 中的
col1#223, col2#224, pt#225 也是从表test_a_pt直接获取到的字段，和表的字段保持一致
col1#226 是从 Project的col1#223 AS col1#226取到的（作为Expand的输入表达式），
null 根据grouping sets的特性而增加的一行值（作为Expand的输入表达式）
1 也是增加的一行值（作为Expand的输入表达式）
List(col1#223, col2#224, pt#225, null, col2#227, 2) 解释也和上面一样，只不过null的位置发生了变化，而1变成了2，这是为了做聚合的时候进行区分

[col1#223, col2#224, pt#225, col1#229, col2#230, spark_grouping_id#228L] 这些是expand的输出expression，其中

col1#223, col2#224, pt#225 和表test_a_pt的字段值一样
col1#229, col2#230, spark_grouping_id#228L 是expand做的的扩展字段，
因为col1和col2的值可能为null，所以exprId和表test_a_pt不一致，
spark_grouping_id#228L 纯属于虚拟字段

而且expand的输入字段是一个Seq(Seq)，这在ExpandExec的时候，会进行row的倍数扩大，Seq里的元素有几个，就会扩展多少倍。

对于Aggregate

Aggregate [col1#229, col2#230, spark_grouping_id#228L], [col1#229, count(1) AS alias#221L]

其中，

[col1#229, col2#230, spark_grouping_id#228L]就是把Expand的输出字段，按照这三个表达式进行group by 聚合
[col1#229, count(1) AS alias#221L] 是聚合表达式，包括聚合的部分字段和部分聚合函数，也就是select语句count(*),col1 as alias

至此Grouping sets 转Expand就分析完了。