Spark同表子查询性能优化：窗口函数与mapPartitions方案探索

Hey folks, let me walk through a tricky performance problem I encountered while working with self-join subqueries using Spark 2.2 and Scala, plus the optimizations we landed on:

The Root Problem: Unnecessary Parquet Disk Scans in Self-Joins #

When we ran queries that involved self-joins on table1 via subqueries, Spark's execution plan ended up triggering repeated full scans of the Parquet files on disk. This excessive IO overhead made our jobs run way slower than expected—total bottleneck situation.

First Fix: Window Functions Replace Simple Self-Joins #

For the simpler self-join scenarios, we refactored the queries using window functions instead, which completely eliminated the duplicate disk reads. For example, instead of using a subquery to fetch related data within the same group, we used partitionBy() to group the data, then combined it with functions like lag(), lead(), or collect_list() to get the needed results in a single pass over the Parquet files. The performance jump was immediately noticeable.

Stuck on the Complex Q2 Query #

Unfortunately, when we moved to the more complex Q2 query, we couldn't find a straightforward way to replace the self-join logic with window functions. The repeated disk scans persisted, and we weren't seeing the performance gains we needed.

Final Optimization: Leveraging mapPartitions for Efficient Processing #

We switched gears to lower-level operator optimization and implemented custom logic using mapPartitions, which solved the problem efficiently:

• First, we generated the fieldcnts table directly within each partition, handling data aggregation and structuring in-memory
• We added the conditional update logic for the type field right in the partition iterator, avoiding unnecessary shuffles and disk IO operations entirely

By processing data within each partition's memory space instead of relying on Spark's high-level join operations, we cut down on disk scans drastically and got the query running at a much faster pace.

内容的提问来源于stack exchange，提问作者Doug T