Apache Spark SQL CLUSTER BY 子句完全指南分区内聚类与排序的语义、实现与实战【免费下载链接】sparkApache Spark - A unified analytics engine for large-scale data processing项目地址: https://gitcode.com/gh_mirrors/sp/sparkCLUSTER BY是 Apache Spark SQL 中用于按表达式重新分区数据并在每个分区内部完成排序的查询组织子句其语义等价于DISTRIBUTE BY加上SORT BY。本文基于当前仓库的官方文档 docs/sql-ref-syntax-qry-select-clusterby.md 与 Catalyst 解析器、逻辑计划源码系统讲解其语法、参数、与ORDER BY/SORT BY/DISTRIBUTE BY的异同、底层实现原理以及可复制的完整示例帮助你正确使用该子句完成数据倾斜预处理、ETL 落盘优化等场景中的分区聚类与局部排序。概述CLUSTER BY 做了什么CLUSTER BY子句执行两步操作重新分区repartition根据输入表达式对数据进行重分布使得具有相同表达式值或其哈希值相同的行被放入同一个分区分区内排序sort在重分区完成后对每个分区内部的数据按照这些表达式进行排序。因此CLUSTER BY在语义上完全等价于先执行 DISTRIBUTE BY负责重分区再执行 SORT BY负责分区内排序。这里有一个关键前提需要牢记CLUSTER BY只保证结果行在每个分区内部是有序的并不保证整个输出是全局有序的。如果需要全局有序的输出必须使用 ORDER BY 子句。语法CLUSTER BY { expression [ , ... ] }支持指定一个或多个表达式多个表达式之间以逗号分隔CLUSTER BY出现在SELECT语句的查询组织query organization部分通常位于WHERE、GROUP BY、HAVING等子句之后。参数说明expression参数说明expression指定一个或多个值、运算符和 SQL 函数组合而成的计算结果。CLUSTER BY既按该表达式的结果对数据进行重分区也按该表达式的值在分区内排序。表达式可以是简单列名如age也可以是任意可求值的表达式例如算术运算、字符串函数、CASE表达式等。在实际查询中它既可以引用SELECT列表中出现的列也可以引用源表中未出现在结果集中的列。完整示例从无排序到分区聚类下面的示例完整取自官方文档通过把 shuffle 分区数降到 2可以更直观地观察CLUSTER BY的聚类与排序行为。CREATE TABLE person (name STRING, age INT); INSERT INTO person VALUES (Zen Hui, 25), (Anil B, 18), (Shone S, 16), (Mike A, 25), (John A, 18), (Jack N, 16); -- 将 shuffle 分区数降为 2便于观察 CLUSTER BY 的聚类与排序行为 SET spark.sql.shuffle.partitions 2;第一步不加任何排序子句的普通查询。没有任何排序指令时查询结果是不确定的age列并未排序SELECT age, name FROM person; ---------- |age| name| ---------- | 16|Shone S| | 25|Zen Hui| | 16| Jack N| | 25| Mike A| | 18| John A| | 18| Anil B| ----------第二步使用CLUSTER BY age。相同age的人员被聚到同一分区且每个分区内部按age升序排列。在示例的输出中年龄为 18 和 25 的人员位于第一个分区年龄为 16 的人员位于第二个分区SELECT age, name FROM person CLUSTER BY age; ---------- |age| name| ---------- | 18| John A| | 18| Anil B| | 25|Zen Hui| | 25| Mike A| | 16|Shone S| | 16| Jack N| ----------对比两个结果可以发现使用CLUSTER BY后相同age值的行必然相邻聚类且每个分区内的行按age有序但整个结果集依然是分区间的部分有序——第一个分区输出18/25后第二个分区才输出16全局来看并非严格升序。等价写法验证DISTRIBUTE BY SORT BY由于CLUSTER BY age语义上等价于先DISTRIBUTE BY age再SORT BY age你完全可以用下面的写法得到等价的结果SELECT age, name FROM person DISTRIBUTE BY age SORT BY age;值得注意的区别是DISTRIBUTE BY只做重分区、不排序其示例输出中分区内行的顺序是随机的SORT BY只做分区内排序、不改变分区方式CLUSTER BY则是两者行为的叠加并且分区内排序方向固定为升序。源码视角CLUSTER BY 在 Catalyst 中如何实现从当前仓库的源码可以确认CLUSTER BY的完整执行路径这有助于理解其重分区 局部排序的底层语义。解析阶段AstBuilder 生成逻辑计划在 sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala 的withQueryResultClauses方法中CLUSTER BY被翻译为如下逻辑计划} else if (order.isEmpty sort.isEmpty distributeBy.isEmpty !clusterBy.isEmpty) { clause PipeOperators.clusterByClause val expressions expressionList(clusterBy) Sort( expressions.map(SortOrder(_, Ascending)), global false, withRepartitionByExpression(ctx, expressions, query)) }这段代码清晰地揭示了三点实现事实重分区算子通过withRepartitionByExpression生成RepartitionByExpression逻辑算子分区内排序通过Sort(..., global false, ...)生成非全局排序global false正是仅在分区内排序的源码级体现排序方向固定每个表达式被包装成SortOrder(_, Ascending)即CLUSTER BY的排序方向固定为升序这一点与SORT BY可显式指定ASC/DESC以及NULLS FIRST/NULLS LAST不同。同时该解析方法还规定ORDER BY、SORT BY、DISTRIBUTE BY、CLUSTER BY是互斥的——如果在一个查询中同时出现多个这类子句会抛出combinationQueryResultClausesUnsupportedError。逻辑计划阶段RepartitionByExpression 算子重分区算子定义在 sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scalacase class RepartitionByExpression( partitionExpressions: Seq[Expression], child: LogicalPlan, optNumPartitions: Option[Int], optAdvisoryPartitionSize: Option[Long] None) extends RepartitionOperation with HasPartitionExpressions { override def shuffle: Boolean true ... }关键点是override def shuffle: Boolean trueCLUSTER BY必然触发一次shuffle即数据重分布这意味着它涉及网络传输与磁盘读写开销与无需 shuffle 的ORDER BY单分区场景除外在代价模型上并不相同。对于大规模数据shuffle 的开销需要通过调整spark.sql.shuffle.partitions等参数来控制并行度与分区大小。物理执行角度RepartitionByExpression在物理执行阶段会转换为基于表达式哈希值HashPartitioning的Exchange算子数据到达各分区后再执行局部的排序。因此可以推断具有相同表达式值的行必然落在同一分区但多个分区之间不存在全局顺序约束这正是文档所述只保证分区内有序、不保证全局有序的实现根源。CLUSTER BY 与 ORDER BY / SORT BY / DISTRIBUTE BY 的对照子句是否重分区shuffle是否分区内排序是否全局排序排序方向控制CLUSTER BY是是否固定升序DISTRIBUTE BY是否否不适用SORT BY否是否可指定ASC/DESC、NULLS FIRST/LASTORDER BY否单分区可无 shuffle是是可指定ASC/DESC、NULLS FIRST/LAST使用建议需要全局有序结果如对外输出必须严格升序→ 使用ORDER BY只想做分区内排序、保持现有分区→ 使用SORT BY只想把相同键的行聚合到同一分区、不关心顺序→ 使用DISTRIBUTE BY既要按键分区聚类、又要每个分区内有序→ 使用CLUSTER BY例如将数据按日期/用户 ID 聚类后再分区写入下游系统可减少下游的局部排序与随机 IO。进阶管道操作符中的 CLUSTER BY当前仓库已支持 SQL 管道操作符pipe operators语法CLUSTER BY也可以作为管道操作符使用。在 sql/core/src/test/scala/org/apache/spark/sql/execution/SparkSqlParserSuite.scala 中存在如下测试checkRepartition(TABLE t | CLUSTER BY x | TABLESAMPLE (100 PERCENT))这表明你可以写出这样的管道式查询TABLE person | CLUSTER BY age;管道操作符语法中| CLUSTER BY与普通SELECT ... CLUSTER BY ...在语义上一致为流式组合多个查询步骤提供了更直观的写法。相关子句速览CLUSTER BY属于SELECT查询组织子句家族与之紧密相关的文档包括SELECT 主语句WHERE ClauseGROUP BY ClauseHAVING ClauseORDER BY ClauseSORT BY ClauseDISTRIBUTE BY ClauseLIMIT ClauseOFFSET ClauseCASE ClausePIVOT ClauseUNPIVOT ClauseLATERAL VIEW Clause小结CLUSTER BY是重分区聚类 分区内升序排序二合一的高性价比子句在数据倾斜预处理、ETL 分区落盘、减少下游局部排序等场景中非常实用。理解其三个要点即可正确使用语义上等于DISTRIBUTE BYSORT BY只保证分区内有序、不保证全局有序必然触发一次 shuffle 且排序方向固定为升序。【免费下载链接】sparkApache Spark - A unified analytics engine for large-scale data processing项目地址: https://gitcode.com/gh_mirrors/sp/spark创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Apache Spark SQL CLUSTER BY 子句完全指南：分区内聚类与排序的语义、实现与实战
Apache Spark SQL CLUSTER BY 子句完全指南分区内聚类与排序的语义、实现与实战【免费下载链接】sparkApache Spark - A unified analytics engine for large-scale data processing项目地址: https://gitcode.com/gh_mirrors/sp/sparkCLUSTER BY是 Apache Spark SQL 中用于按表达式重新分区数据并在每个分区内部完成排序的查询组织子句其语义等价于DISTRIBUTE BY加上SORT BY。本文基于当前仓库的官方文档 docs/sql-ref-syntax-qry-select-clusterby.md 与 Catalyst 解析器、逻辑计划源码系统讲解其语法、参数、与ORDER BY/SORT BY/DISTRIBUTE BY的异同、底层实现原理以及可复制的完整示例帮助你正确使用该子句完成数据倾斜预处理、ETL 落盘优化等场景中的分区聚类与局部排序。概述CLUSTER BY 做了什么CLUSTER BY子句执行两步操作重新分区repartition根据输入表达式对数据进行重分布使得具有相同表达式值或其哈希值相同的行被放入同一个分区分区内排序sort在重分区完成后对每个分区内部的数据按照这些表达式进行排序。因此CLUSTER BY在语义上完全等价于先执行 DISTRIBUTE BY负责重分区再执行 SORT BY负责分区内排序。这里有一个关键前提需要牢记CLUSTER BY只保证结果行在每个分区内部是有序的并不保证整个输出是全局有序的。如果需要全局有序的输出必须使用 ORDER BY 子句。语法CLUSTER BY { expression [ , ... ] }支持指定一个或多个表达式多个表达式之间以逗号分隔CLUSTER BY出现在SELECT语句的查询组织query organization部分通常位于WHERE、GROUP BY、HAVING等子句之后。参数说明expression参数说明expression指定一个或多个值、运算符和 SQL 函数组合而成的计算结果。CLUSTER BY既按该表达式的结果对数据进行重分区也按该表达式的值在分区内排序。表达式可以是简单列名如age也可以是任意可求值的表达式例如算术运算、字符串函数、CASE表达式等。在实际查询中它既可以引用SELECT列表中出现的列也可以引用源表中未出现在结果集中的列。完整示例从无排序到分区聚类下面的示例完整取自官方文档通过把 shuffle 分区数降到 2可以更直观地观察CLUSTER BY的聚类与排序行为。CREATE TABLE person (name STRING, age INT); INSERT INTO person VALUES (Zen Hui, 25), (Anil B, 18), (Shone S, 16), (Mike A, 25), (John A, 18), (Jack N, 16); -- 将 shuffle 分区数降为 2便于观察 CLUSTER BY 的聚类与排序行为 SET spark.sql.shuffle.partitions 2;第一步不加任何排序子句的普通查询。没有任何排序指令时查询结果是不确定的age列并未排序SELECT age, name FROM person; ---------- |age| name| ---------- | 16|Shone S| | 25|Zen Hui| | 16| Jack N| | 25| Mike A| | 18| John A| | 18| Anil B| ----------第二步使用CLUSTER BY age。相同age的人员被聚到同一分区且每个分区内部按age升序排列。在示例的输出中年龄为 18 和 25 的人员位于第一个分区年龄为 16 的人员位于第二个分区SELECT age, name FROM person CLUSTER BY age; ---------- |age| name| ---------- | 18| John A| | 18| Anil B| | 25|Zen Hui| | 25| Mike A| | 16|Shone S| | 16| Jack N| ----------对比两个结果可以发现使用CLUSTER BY后相同age值的行必然相邻聚类且每个分区内的行按age有序但整个结果集依然是分区间的部分有序——第一个分区输出18/25后第二个分区才输出16全局来看并非严格升序。等价写法验证DISTRIBUTE BY SORT BY由于CLUSTER BY age语义上等价于先DISTRIBUTE BY age再SORT BY age你完全可以用下面的写法得到等价的结果SELECT age, name FROM person DISTRIBUTE BY age SORT BY age;值得注意的区别是DISTRIBUTE BY只做重分区、不排序其示例输出中分区内行的顺序是随机的SORT BY只做分区内排序、不改变分区方式CLUSTER BY则是两者行为的叠加并且分区内排序方向固定为升序。源码视角CLUSTER BY 在 Catalyst 中如何实现从当前仓库的源码可以确认CLUSTER BY的完整执行路径这有助于理解其重分区 局部排序的底层语义。解析阶段AstBuilder 生成逻辑计划在 sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala 的withQueryResultClauses方法中CLUSTER BY被翻译为如下逻辑计划} else if (order.isEmpty sort.isEmpty distributeBy.isEmpty !clusterBy.isEmpty) { clause PipeOperators.clusterByClause val expressions expressionList(clusterBy) Sort( expressions.map(SortOrder(_, Ascending)), global false, withRepartitionByExpression(ctx, expressions, query)) }这段代码清晰地揭示了三点实现事实重分区算子通过withRepartitionByExpression生成RepartitionByExpression逻辑算子分区内排序通过Sort(..., global false, ...)生成非全局排序global false正是仅在分区内排序的源码级体现排序方向固定每个表达式被包装成SortOrder(_, Ascending)即CLUSTER BY的排序方向固定为升序这一点与SORT BY可显式指定ASC/DESC以及NULLS FIRST/NULLS LAST不同。同时该解析方法还规定ORDER BY、SORT BY、DISTRIBUTE BY、CLUSTER BY是互斥的——如果在一个查询中同时出现多个这类子句会抛出combinationQueryResultClausesUnsupportedError。逻辑计划阶段RepartitionByExpression 算子重分区算子定义在 sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scalacase class RepartitionByExpression( partitionExpressions: Seq[Expression], child: LogicalPlan, optNumPartitions: Option[Int], optAdvisoryPartitionSize: Option[Long] None) extends RepartitionOperation with HasPartitionExpressions { override def shuffle: Boolean true ... }关键点是override def shuffle: Boolean trueCLUSTER BY必然触发一次shuffle即数据重分布这意味着它涉及网络传输与磁盘读写开销与无需 shuffle 的ORDER BY单分区场景除外在代价模型上并不相同。对于大规模数据shuffle 的开销需要通过调整spark.sql.shuffle.partitions等参数来控制并行度与分区大小。物理执行角度RepartitionByExpression在物理执行阶段会转换为基于表达式哈希值HashPartitioning的Exchange算子数据到达各分区后再执行局部的排序。因此可以推断具有相同表达式值的行必然落在同一分区但多个分区之间不存在全局顺序约束这正是文档所述只保证分区内有序、不保证全局有序的实现根源。CLUSTER BY 与 ORDER BY / SORT BY / DISTRIBUTE BY 的对照子句是否重分区shuffle是否分区内排序是否全局排序排序方向控制CLUSTER BY是是否固定升序DISTRIBUTE BY是否否不适用SORT BY否是否可指定ASC/DESC、NULLS FIRST/LASTORDER BY否单分区可无 shuffle是是可指定ASC/DESC、NULLS FIRST/LAST使用建议需要全局有序结果如对外输出必须严格升序→ 使用ORDER BY只想做分区内排序、保持现有分区→ 使用SORT BY只想把相同键的行聚合到同一分区、不关心顺序→ 使用DISTRIBUTE BY既要按键分区聚类、又要每个分区内有序→ 使用CLUSTER BY例如将数据按日期/用户 ID 聚类后再分区写入下游系统可减少下游的局部排序与随机 IO。进阶管道操作符中的 CLUSTER BY当前仓库已支持 SQL 管道操作符pipe operators语法CLUSTER BY也可以作为管道操作符使用。在 sql/core/src/test/scala/org/apache/spark/sql/execution/SparkSqlParserSuite.scala 中存在如下测试checkRepartition(TABLE t | CLUSTER BY x | TABLESAMPLE (100 PERCENT))这表明你可以写出这样的管道式查询TABLE person | CLUSTER BY age;管道操作符语法中| CLUSTER BY与普通SELECT ... CLUSTER BY ...在语义上一致为流式组合多个查询步骤提供了更直观的写法。相关子句速览CLUSTER BY属于SELECT查询组织子句家族与之紧密相关的文档包括SELECT 主语句WHERE ClauseGROUP BY ClauseHAVING ClauseORDER BY ClauseSORT BY ClauseDISTRIBUTE BY ClauseLIMIT ClauseOFFSET ClauseCASE ClausePIVOT ClauseUNPIVOT ClauseLATERAL VIEW Clause小结CLUSTER BY是重分区聚类 分区内升序排序二合一的高性价比子句在数据倾斜预处理、ETL 分区落盘、减少下游局部排序等场景中非常实用。理解其三个要点即可正确使用语义上等于DISTRIBUTE BYSORT BY只保证分区内有序、不保证全局有序必然触发一次 shuffle 且排序方向固定为升序。【免费下载链接】sparkApache Spark - A unified analytics engine for large-scale data processing项目地址: https://gitcode.com/gh_mirrors/sp/spark创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考