Spark缓存表查询变慢：ColumnarToRow与RowToColumnar转换疑问

问题核心 #

执行聚合查询SELECT user_id,count(*) FROM transaction GROUP BY 1时：

• 直接查询Parquet源表耗时20秒；
• 通过CACHE TABLE transaction OPTIONS ('storageLevel' 'MEMORY_AND_DISK') as SELECT * FROM transaction缓存表后，相同查询耗时达2分钟，查询计划中新增了ColumnarToRow和RowToColumnar两次转换操作，这是导致性能骤降的核心瓶颈。

原因解析 #

1. 缓存格式与数据源不匹配 #

Spark SQL默认的CACHE TABLE操作会将数据缓存为行式存储的RDD格式，而Parquet是列存格式。当执行聚合这类列导向的操作时：

• 首先需要将行式缓存数据转换为列格式（RowToColumnar），才能高效执行按列聚合的逻辑；
• 聚合完成后，又需要将列格式结果转换回行格式（ColumnarToRow），以适配Spark后续的行式数据处理流程。
  两次转换带来了大量额外的CPU和内存开销，直接拖慢了查询速度。

2. 直接读取Parquet的原生优化 #

直接查询Parquet表时，Spark的Parquet数据源会自动执行列裁剪（仅读取查询所需的user_id列），并且全程以列格式处理聚合操作，无需进行任何行列转换，因此效率更高。

3. 两次反向转换的本质 #

行→列转换是为了利用列存格式在聚合计算上的效率优势；列→行转换则是因为Spark的上层算子（如结果输出）默认基于行式数据模型。而直接读取Parquet时，数据源直接提供列数据，聚合后可直接输出，自然跳过了这两次转换。

解决方案 #

• 改用DataFrame/Dataset API缓存（推荐）
  避免使用SQL的CACHE TABLE语句，改用DataFrame的缓存API，Spark会自动适配Parquet的列存特性，将数据缓存为列式格式：

  // 读取Parquet表
  val transactionDF = spark.read.parquet("path/to/transaction")
  // 缓存数据（可指定存储级别）
  transactionDF.persist(org.apache.spark.storage.StorageLevel.MEMORY_AND_DISK)
  // 创建临时视图供SQL查询
  transactionDF.createOrReplaceTempView("transaction")
  

• 配置SQL缓存为列序列化格式
  通过Spark配置指定缓存序列化器为列式序列化器，这样使用CACHE TABLE时也会以列格式缓存数据：

  # 提交任务时指定配置
  spark-submit --conf spark.sql.cache.serializer=org.apache.spark.sql.execution.columnar.ColumnarBatchSerializer ...
  # 或在spark-defaults.conf中持久化配置
  spark.sql.cache.serializer org.apache.spark.sql.execution.columnar.ColumnarBatchSerializer
  

• 按需缓存必要列，缩减数据量
  如果你的查询仅依赖user_id列，缓存时只保留需要的列，即使存在转换，数据量也会大幅降低：

  CACHE TABLE transaction_cache OPTIONS ('storageLevel' 'MEMORY_AND_DISK') AS SELECT user_id FROM transaction;
  -- 查询时使用精简后的缓存表
  SELECT user_id, count(*) FROM transaction_cache GROUP BY 1;
  

补充说明 #

日常处理Parquet表很少看到ColumnarToRow/RowToColumnar转换，是因为Spark对Parquet数据源做了端到端的列优化：从列数据读取、列计算到结果输出，全程尽量维持列格式，避免不必要的行列转换。只有当数据被强制缓存为行式格式时，才会触发这类额外的转换操作。

内容的提问来源于stack exchange，提问作者Eugenio.Gastelum96