当前位置：首页 > news >正文

Spark专题-第二部分：Spark SQL 入门（8）-算子介绍-sort

news 2025/9/28 5:15:43

Spark专题-第二部分：Spark SQL 入门（8）-算子介绍-sort

要不是在工作中发现了性能极差的cluster by写法，真把sort算子给忘了，赶紧补上

Sort算子概述

在Spark SQL中，排序操作通过SortExec物理算子实现。不同的SQL排序语法会产生不同的执行计划，涉及不同的数据分布和排序策略。

排序相关SQL语法及对应算子

1. ORDER BY

描述: 全局排序，对整个数据集进行排序
对应算子: SortExec + Exchange（单分区）
执行特点: 需要将所有数据收集到单个Executor进行排序

SQL写法:

SELECT * FROM sales ORDER BY sale_amount DESC;

2. SORT BY

描述: 在每个分区内排序，不保证全局有序
对应算子: SortExec（无Exchange）
执行特点: 分区内排序，性能较好但结果不是全局有序

SQL写法:

SELECT * FROM sales SORT BY sale_amount DESC;

3. DISTRIBUTE BY

描述: 按指定列重新分布数据，但不排序
对应算子: Exchange（哈希分区）
执行特点: 仅重新分区，不进行排序

SQL写法:

SELECT * FROM sales DISTRIBUTE BY department;

4. CLUSTER BY

描述: DISTRIBUTE BY和SORT BY的组合，按相同列分区和排序
对应算子: Exchange + SortExec
执行特点: 先按列分区，然后在每个分区内按同一列排序

SQL写法:

SELECT * FROM sales CLUSTER BY department;

物理执行计划对比

ORDER BY执行流程

SORT BY执行流程

DISTRIBUTE BY执行流程

CLUSTER BY执行流程

实际案例与执行计划分析

案例: 销售数据分析

-- 创建示例表
CREATE TABLE sales (sale_id INT,salesperson STRING,department STRING,sale_amount DOUBLE,sale_date DATE
);-- 插入数据
INSERT INTO sales VALUES
(1, 'Alice', 'Tech', 1000.0, '2023-01-15'),
(2, 'Bob', 'Tech', 1500.0, '2023-01-16'),
(3, 'Alice', 'Tech', 800.0, '2023-02-10'),
(4, 'Charlie', 'HR', 1200.0, '2023-01-20'),
(5, 'Bob', 'Tech', 2000.0, '2023-02-05');

ORDER BY执行计划

EXPLAIN EXTENDED
SELECT * FROM sales ORDER BY sale_amount DESC;

物理执行计划:

== Physical Plan ==
*(2) Sort [sale_amount#10 DESC NULLS LAST], true, 0
+- Exchange SinglePartition, ENSURE_REQUIREMENTS, [id=#15]
+- *(1) Scan ExistingRDD[sale_id#8,salesperson#9,sale_amount#10,department#11,sale_date#12]

SORT BY执行计划

EXPLAIN EXTENDED
SELECT * FROM sales SORT BY sale_amount DESC;

物理执行计划:
== Physical Plan ==
*(1) Sort [sale_amount#10 DESC NULLS LAST], false, 0
± *(1) Scan ExistingRDD[sale_id#8,salesperson#9,sale_amount#10,department#11,sale_date#12]

DISTRIBUTE BY + SORT BY执行计划

EXPLAIN EXTENDED
SELECT * FROM sales 
DISTRIBUTE BY department 
SORT BY sale_amount DESC;

物理执行计划:

== Physical Plan ==
*(2) Sort [sale_amount#10 DESC NULLS LAST], false, 0
+- Exchange hashpartitioning(department#11, 200), ENSURE_REQUIREMENTS, [id=#15]
+- *(1) Scan ExistingRDD[sale_id#8,salesperson#9,sale_amount#10,department#11,sale_date#12]

性能优化策略

1. 避免全局排序（ORDER BY）

2. 分区策略优化

-- 调整分区数避免数据倾斜
SET spark.sql.shuffle.partitions=100;-- 对于已知数据分布，使用范围分区
SET spark.sql.adaptive.enabled=true;
SET spark.sql.adaptive.coalescePartitions.enabled=true;

3. 内存管理优化

-- 调整排序内存
SET spark.sql.sort.spill.numElementsForceSpillThreshold=1000000;
SET spark.sql.execution.sort.spill.initialMemoryThreshold=100000;-- 启用外部排序
SET spark.sql.sort.enableRadixSort=true;
SET spark.sql.useExternalSort=true;

不同场景下的选择策略

场景1: 生成排序报告（需要全局有序）

-- 小数据量：使用ORDER BY
SELECT * FROM daily_sales ORDER BY sale_amount DESC;-- 大数据量：分阶段处理
CREATE TABLE temp_sorted AS
SELECT * FROM big_sales SORT BY sale_amount DESC;-- 然后对较小结果集进行全局排序（如果需要）
SELECT * FROM temp_sorted ORDER BY sale_amount DESC;

场景2: 为后续聚合操作准备数据

-- 使用CLUSTER BY优化后续的窗口函数
SELECT salesperson,SUM(sale_amount) OVER (PARTITION BY department ORDER BY sale_date) 
FROM sales 
CLUSTER BY department, sale_date;

场景3: 数据重新分布

-- 使用DISTRIBUTE BY优化数据分布
INSERT OVERWRITE TABLE sales_by_dept
SELECT * FROM sales DISTRIBUTE BY department;

执行计划解析技巧

识别排序类型

在物理执行计划中关注：

全局排序: Sort [column ASC/DESC], true, 0 + Exchange SinglePartition
分区内排序: Sort [column ASC/DESC], false, 0
分区排序: Sort [column ASC/DESC], false, 0 + Exchange hashpartitioning

性能瓶颈诊断

-- 查看排序操作的详细统计
EXPLAIN COST
SELECT * FROM sales ORDER BY sale_amount DESC;-- 监控排序过程中的数据倾斜
SET spark.sql.adaptive.skew.enabled=true;
SET spark.sql.adaptive.logLevel=DEBUG;