Spark SQL 查询计划中的 ShuffleQueryStage 和 ReusedQueryStage答案

【问题标题】：ShuffleQueryStage and ReusedQueryStage in Spark SQL query PlansSpark SQL 查询计划中的 ShuffleQueryStage 和 ReusedQueryStage
【发布时间】：2022-12-16 03:16:28
【问题描述】：

ShuffleQueryStage 20 和 ReusedQueryStage 16 在下面的 Spark SQL 查询计划中是什么意思？我已经分享了为我的查询生成的查询计划的一部分。

我正在使用星火 2.4.7。

:     +- ReusedQueryStage 16
:        +- BroadcastQueryStage 7
:           +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, string, true]))
:              +- AdaptiveSparkPlan(isFinalPlan=true)
:                 +- *(11) HashAggregate(keys=[src_clmorigid#21055], functions=[], output=[src_clmorigid#21055])
:                    +- ShuffleQueryStage 21, true
:                       +- Exchange hashpartitioning(src_clmorigid#21055, 10)
:                          +- *(10) HashAggregate(keys=[src_clmorigid#21055], functions=[], output=[src_clmorigid#21055])
:                             +- *(10) Project [src_clmorigid#21055]
:                                +- *(10) BroadcastHashJoin [tgt_clmorigid#21152], [tgt_clmorigid#20756], Inner, BuildRight
:                                   :- *(10) Project [src_clmorigid#21055, tgt_clmorigid#21152]
:                                   :  +- *(10) Filter (isnotnull(tgt_clmorigid#21152) && isnotnull(src_clmorigid#21055))
:                                   :     +- *(10) FileScan parquet default.vw_exclude_latest_set_frm_clm[src_clmorigid#21055,tgt_clmorigid#21152] Batched: true, Format: Parquet, Location: InMemoryFileIndex[s3://dm_bucket...
:                                   +- ReusedQueryStage 20
:                                      +- BroadcastQueryStage 6
:                                         +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, string, true]))
:                                            +- AdaptiveSparkPlan(isFinalPlan=true)
:                                               +- *(9) HashAggregate(keys=[tgt_clmorigid#20756], functions=[], output=[tgt_clmorigid#20756])
:                                                  +- ShuffleQueryStage 19, true
:                                                     +- Exchange hashpartitioning(tgt_clmorigid#20756, 10)
:                                                        +- *(8) HashAggregate(keys=[tgt_clmorigid#20756], functions=[], output=[tgt_clmorigid#20756])
:                                                           +- *(8) Project [tgt_clmorigid#20756]
:                                                              +- *(8) Filter ((((isnotnull(tgt_clm_line_type_ind#20783) && isnotnull(src_clm_line_type_ind#20686)) 
:                                                                 +- *(8) FileScan parquet default.vw_exclude_latest_set_frm_clm[src_clm_line_type_ind#20686,tgt_clmorigid#20756,tgt_clm_line_type_ind#20783] Batched: true, Format: Parquet, Location: InMemoryFileIndex[s3://...PushedFilters: [IsNotNull(tgt_clm_line_type_ind), 
                                                                      +- *(41) Project [vw_clm_base_fact_sk#21807, source_system#21808, eff_date#21809, frst_sales_crtn_dt#21810, clmorigid#21811, ... 59 more fields]
                                                                          +- *(41) FileScan parquet default.vw_to_be_merged_data[vw_clm_base_fact_sk#21807,source_system#21808,eff_date#21809,frst_sales_crtn_dt#21810,... 56 more fields], ...

如果需要，很乐意提供更多信息。

【问题讨论】：

标签： apache-spark-sql

【解决方案1】：

此阶段连接到 AQE（自适应查询执行）。如果您使用 AQE 执行代码，它们应该会消失

ShuffleQueryStage - 这是在交换之后添加的，它用于具体化前一阶段的结果（因此交换）以允许 AQE 使用运行时统计信息并重新优化计划

ReusedQueryStage - 这意味着该执行分支已经在您的计划中并且可以重复使用。在源代码中，我发现一条评论说查询顺序可能不直观

我知道这听起来可能很奇怪，因为 AQE 是在 Spark 3 中正式发布的，但实际上它可以从 Spark 2.2.x 获得（与现在工作的形状不同）所以可能有人在你的集群上使用 Spark 2.4.7 和启用了 AQE你看到这个阶段

我在其他答案中添加了 ShuffleQueryStage 分析示例，您可以看一下：Other answer

【讨论】：