Spark随机采样速度过慢求助：1亿条数据采样耗时约5小时

算力阿峰

2026-5-25

问题分析与优化方案

先结合你的代码和场景拆解下耗时超预期的核心原因，再给出针对性的优化建议：

1. 采样逻辑的明显矛盾：比例设置与Limit的冗余

你设置的frac = 1e-5，对应1亿条记录的采样量仅为1000条（100e6 × 0.00001），但后续又加了limit(nSamples)（30万条）。这会带来两个问题：

实际采样结果远达不到你想要的30万条，完全不符合业务预期；
Spark会先完成全表sample操作，再执行limit，但因为sample结果远小于limit值，这个limit完全是多余的，还会增加不必要的阶段开销。

修正方式：先计算正确的采样比例：

val totalCount = spark.table("db_name.table_name").count() // 1亿条记录的count操作耗时很短
val nSamples = 3e5.toInt
val correctFrac = nSamples.toDouble / totalCount // 即3e5/1e8 = 0.003

2. 小分区过多导致调度开销爆炸

你的源表有11433个分区，对应1亿条记录，平均每个分区仅约8750条数据。Spark中每个分区对应一个任务，这么多小任务会直接拉高整体耗时：

大量任务的调度、启动、销毁开销，可能远超过数据本身的处理时间；
集群资源无法充分利用：6个Worker节点就算每个配8核，总核数也才48，同时只能跑48个任务，剩下的1万多个任务需要排队等待调度。

优化方式：合并小分区，让每个分区的数据量更合理（建议每个分区50万-100万条，对应1亿条记录设置100-200个分区即可）：

// 用coalesce（无shuffle，适合数据分布均匀的场景）或repartition（有shuffle，数据更均匀）合并分区
val optimizedTable = spark.table("db_name.table_name")
  .coalesce(200) // 可根据集群总核数调整，比如总核数48的话，设置96-200都合适

3. 调整后的完整采样代码

结合上面的修正，最终的优化代码如下：

import org.apache.spark.sql.SaveMode

val nSamples = 3e5.toInt
val totalCount = spark.table("db_name.table_name").count()
val correctFrac = nSamples.toDouble / totalCount

spark.table("db_name.table_name")
  .coalesce(200)
  .sample(false, correctFrac, seed = 42) // seed保证采样结果可重复
  .write
  .mode(SaveMode.Overwrite)
  .saveAsTable("db_name.new_name")

如果不想先执行count操作，也可以用randomSplit直接按比例拆分：

val Array(sampleDF, _) = spark.table("db_name.table_name")
  .coalesce(200)
  .randomSplit(Array(correctFrac, 1 - correctFrac), seed = 42)

sampleDF.write
  .mode(SaveMode.Overwrite)
  .saveAsTable("db_name.new_name")

4. 集群配置的额外调优建议

针对你的6个Worker节点，还可以调整Spark配置进一步提升效率：

设置spark.executor.cores：每个executor分配合适的核心数，比如每个Worker给4核，总核数24；
设置spark.sql.shuffle.partitions：如果用到repartition（带shuffle），将该值设为目标分区数（比如200），避免默认2000个小分区的开销；
适当增加executor内存：减少GC频率，提升数据处理速度。

按照这些优化点调整后，采样耗时应该会大幅降低，从5小时压缩到几十分钟甚至更短。

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