Spark简单查询优化：百万级多列数据表性能提升咨询

问题背景 #

我有一个用于生成count(*)、min()、max()及列类型过滤count的SQL查询，示例DataFrame如下：

id  val  desc   date
1   10   Test   2023-01-01
2   20   Test1  2023-10-01
3   30          2023-01-02
4   40   Test2  2023-01-01

针对该DataFrame，脚本生成的查询如下：

SELECT 
COUNT(DISTINCT id) as distinct_id,
COUNT(*) as rows_id,
COUNT(DISTINCT id) as distinct_val,
COUNT(*) as rows_val,
COUNT(DISTINCT desc) as distinct_desc,
COUNT(*) as rows_desc,
(SELECT COUNT(*) FROM table WHERE TRIM(desc) = '') as missing_desc,
COUNT(DISTINCT date) as distinct_date,
COUNT(*) as rows_date,
COUNT(date) as count_date,
MIN(date) as min_date,
MAX(date) as max_date
FROM table

生产环境中表有约100万行、150列，结果DataFrame至少有350列。通过df.explain()查看查询计划，发现每个列对应的查询逻辑重复扫描表，导致性能极差，提取指标耗时约10分钟（集群初始1个Worker，支持自动扩容至20个）。

现咨询两个问题：

1. 不改变当前查询结构的前提下，是否可通过Spark额外配置优化执行时间？
2. 若需修改查询，获取各列对应指标的最优方式是什么？

备注：已尝试df.summary()，但无法提供日期类型指标或字符串自定义指标，因此自行编写SQL查询。

问题1：不修改查询结构的Spark配置优化 #

可以通过以下Spark配置调整缓解重复扫描带来的性能问题：

• 开启表缓存：
  • 设置spark.sql.cache.serialized=true，缓存时序列化数据，减少内存占用并提升缓存效率；
  • 执行CACHE TABLE table提前将表缓存到内存，避免后续重复扫描磁盘。
• 调整并行度：
  • 设置spark.sql.shuffle.partitions=400（根据20个Worker的规模调整，一般建议每个Executor处理2-4个分区），避免shuffle阶段出现小分区或大分区瓶颈；
  • 设置spark.default.parallelism=80（按每个Worker分配4个核心计算，20*4=80），匹配集群计算资源。
• 启用动态分区裁剪：
  • 设置spark.sql.optimizer.dynamicPartitionPruning.enabled=true，优化子查询的过滤逻辑，减少不必要的数据扫描。
• 优化内存配置：
  • 调大spark.executor.memory=8g和spark.driver.memory=4g，避免内存不足导致的磁盘落盘操作，提升计算效率。
• 启用向量化读取：
  • 如果表数据格式是列式存储（如Parquet、ORC），设置spark.sql.parquet.enableVectorizedReader=true，提升数据读取速度。

问题2：修改查询的最优方式 #

核心是实现一次全表扫描完成所有统计指标计算，彻底消除重复扫描，以下是两种可行方案：

方案1：使用Spark DataFrame API（推荐，适配多列场景） #

利用agg方法一次性聚合所有指标，Spark会自动生成单一扫描计划。针对多列场景，可以通过遍历列信息自动生成对应指标：

import org.apache.spark.sql.functions._
import org.apache.spark.sql.types._

// 遍历所有列，根据列类型自动生成统计指标
val allMetrics = df.columns.flatMap { colName =>
  val colType = df.schema(colName).dataType
  // 所有列通用的基础指标
  val baseMetrics = Seq(
    countDistinct(colName).alias(s"distinct_$colName"),
    count(lit(1)).alias(s"rows_$colName")
  )
  // 不同数据类型的专属指标
  val typeSpecific = colType match {
    case StringType => 
      Seq(sum(when(trim(col(colName)) === "", 1).otherwise(0)).alias(s"missing_$colName"))
    case DateType | TimestampType => 
      Seq(
        count(colName).alias(s"count_$colName"),
        min(colName).alias(s"min_$colName"),
        max(colName).alias(s"max_$colName")
      )
    case IntegerType | LongType | DoubleType => 
      Seq(
        count(colName).alias(s"count_$colName"),
        min(colName).alias(s"min_$colName"),
        max(colName).alias(s"max_$colName"),
        avg(colName).alias(s"avg_$colName")
      )
    case _ => Seq.empty
  }
  baseMetrics ++ typeSpecific
}

// 执行聚合计算
val resultDf = df.agg(allMetrics: _*)
resultDf.show()

方案2：优化SQL查询（替换子查询为条件聚合） #

将原来的子查询替换为SUM(CASE...)条件聚合，让所有统计逻辑在一次全表扫描中完成：

SELECT 
  COUNT(DISTINCT id) as distinct_id,
  COUNT(*) as rows_id,
  COUNT(DISTINCT val) as distinct_val,
  COUNT(*) as rows_val,
  COUNT(DISTINCT desc) as distinct_desc,
  COUNT(*) as rows_desc,
  SUM(CASE WHEN TRIM(desc) = '' THEN 1 ELSE 0 END) as missing_desc,
  COUNT(DISTINCT date) as distinct_date,
  COUNT(*) as rows_date,
  COUNT(date) as count_date,
  MIN(date) as min_date,
  MAX(date) as max_date
FROM table

原查询中的子查询会触发独立的全表扫描，替换为条件聚合后，所有计算会合并到同一个聚合阶段，只扫描一次表。

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