笔者最近在做一些任务的优化，大多的场景是因为采用row_number()进行分组去重，所以耗时特别高。样例代码如下：
select *
from (
select *
from (
select *,
row_number() over(partition by id order by ts desc) rn
from table
) as a
) as a
where rn = 1
原因
要做这个操作，不得不做的就是shuffle，而且因为要保留每行数据，没法在map端做合并，所以造成shuffle的量特别大。
优化思路
减少shuffle
map端合并
假如没有这样的窗口操作的算子，我们自己来实现一个这样的算子的话，其实是可以在map端做数据合并的。
例如，我们可以把rn=1的操作前置，map端根据分区key和ts排序之后只保留1条就可以，然后reduce端把多个分区的1条记录合并再得到最终的一条。
目前Spark并没有这样的算子，但是可以用其他算子模仿一下，比如max算子，示例代码如下：
select id, split(tuple, ',') arr
from (
select id, max(concat_ws(',', ts, col1, col2, ..)) tuple
from table
group by id
) as a
在笔者的场景中，确实是可以减少shuffle的数据量，但是下一个stage会很耗时。所以这个是一个策略，但是需要按照自己的情况测试下，也要结合数据量。
如果重复的数据不多，这种策略并不是很好。
复用分区
这个是笔者偶然间发现的一个场景。
笔者的业务SQL如下：
select a.*, b.col
from (
select *, row_number() over (partition by id, fid order by client_time desc) rn
from table
) as a
left join mapping_tb as b
on a.id = b.id
where rn = 1
业务逻辑：
分区去重+关联维度表
执行计划如下：
+- SortMergeJoin LeftOuter (18)
:- Sort (10)
: +- Exchange (9)
: +- Filter (7)
: +- Window (6)
: +- Sort (5)
: +- ShuffleQueryStage (4)
: +- Exchange (3)
: +- * ColumnarToRow (2)
: +- Scan orc table (1)
+- Sort (17)
+- ShuffleQueryStage (16)
+- Exchange (15)
+- * Project (14)
+- * Filter (13)
+- * ColumnarToRow (12)
+- Scan orc mapping_tb (11)
可以看到关联维度表用的key为id，分区去重用的key是id+fid，前者的分区肯定包含后者，是不是可以在这上面做些文章。
带着这样的怀疑，对逻辑进行了调整，如下：
select *
from (
select *, row_number() over (partition by id, fid order by client_time desc) rn
from (
select a.*, b.col
from table as a
left join mapping_tb as b
on a.id = b.id
) as a
) as a
where rn = 1
执行计划如下：
+- Filter (17)
+- Window (16)
+- Sort (15)
+- Project (14)
+- SortMergeJoin LeftOuter (13)
:- Sort (5)
: +- ShuffleQueryStage (4)
: +- Exchange (3)
: +- * ColumnarToRow (2)
: +- Scan orc table (1)
+- Sort (12)
+- ShuffleQueryStage (11)
+- Exchange (10)
+- * Project (9)
+- * Filter (8)
+- * ColumnarToRow (7)
+- Scan orc mapping_tb (6)
从这里可以看出SortMergeJoin之后做窗口操作，并没有做Shuffle，而是直接接了一个Sort操作，这就是复用分区。
出于对比我们将上述的SQL做调整并比较执行计划。
select *
from (
select *, row_number() over (partition by id2, fid order by client_time desc) rn
from (
select a.*, b.col
from table as a
left join mapping_tb as b
on a.id = b.id
) as a
) as a
where rn = 1
执行计划如下：
+- Filter (18)
+- Window (17)
+- Sort (16)
+- Exchange (15)
+- Project (14)
+- SortMergeJoin LeftOuter (13)
:- Sort (5)
: +- ShuffleQueryStage (4)
: +- Exchange (3)
: +- * ColumnarToRow (2)
: +- Scan orc table (1)
+- Sort (12)
+- ShuffleQueryStage (11)
+- Exchange (10)
+- * Project (9)
+- * Filter (8)
+- * ColumnarToRow (7)
+- Scan orc mapping_tb (6)
可以看到，修改之后的SQL执行计划多了一个Exchange (15)，说明这里需要再做一次Shuffle。
实际在执行过程中，Shuffle采用了id做分区，Shuffle的数据量能减少大概1倍，而且后续没有再次的shuffle，即我们将shuffle的数据量减少了1倍。
