Spark on Dataproc向BigQuery插入3000万条数据过慢问题咨询

为什么拆分插入耗时远高于合并方式？ #

核心原因是Spark的惰性求值机制导致上游计算被重复执行：

• 当你将(Reconresult, Reconsource)类型的Dataset拆分为两个DataFrame并分别写入BQ时，每个write操作都是一个Spark Action，会触发完整的DAG计算——从多表关联到map转换的全流程会被执行两次。3000万条记录的上游计算本身就耗时，重复执行自然会让总时间翻倍甚至更多。
• 而合并成单个数据集写入时，整个上游流程只执行一次，后续仅做一次写入处理，所以总耗时大幅降低。

针对大数据集的高效处理方式 #

1. 缓存中间结果 #

将生成(Reconresult, Reconsource)的Dataset提前缓存，避免重复计算上游流程：

val combinedDs = // 生成(Reconresult, Reconsource)的逻辑代码
// 选择MEMORY_AND_DISK_SER存储级别，适合大数据量，避免内存溢出
combinedDs.persist(org.apache.spark.storage.StorageLevel.MEMORY_AND_DISK_SER)

// 拆分后写入BQ
val resultDf = combinedDs.select("reconresult.*").as[Reconresult].toDF()
val sourceDf = combinedDs.select("reconsource.*").as[Reconsource].toDF()

resultDf.write.format("bigquery").option("table", "project.dataset.result_table").save()
sourceDf.write.format("bigquery").option("table", "project.dataset.source_table").save()

// 最后释放缓存
combinedDs.unpersist()

2. 物化中间结果到临时存储 #

如果数据集过大，缓存内存压力高，可以先将中间结果写入GCS的Parquet文件（列式存储，读写高效），再读取拆分写入BQ：

val combinedDs = // 生成(Reconresult, Reconsource)的逻辑代码
// 写入GCS临时路径
combinedDs.write.mode("overwrite").parquet("gs://your-bucket/tmp/combined_data")

// 读取拆分后写入BQ
val combinedDf = spark.read.parquet("gs://your-bucket/tmp/combined_data")
val resultDf = combinedDf.select("reconresult.*")
val sourceDf = combinedDf.select("reconsource.*")

resultDf.write.format("bigquery").option("table", "project.dataset.result_table").save()
sourceDf.write.format("bigquery").option("table", "project.dataset.source_table").save()

3. 优化BQ写入配置 #

• 调整并行度：设置spark.sql.shuffle.partitions（默认200，3000万条可设为500-1000），确保写入BQ时有足够的并行任务，避免单分区写入瓶颈。
• 使用Direct Write模式：Spark BigQuery Connector支持直接写入BQ，无需经过GCS临时文件，减少IO开销：

  resultDf.write.format("bigquery")
    .option("table", "project.dataset.result_table")
    .option("writeMethod", "direct")
    .save()
  

• 开启批量插入：通过option("batchSize", "10000")设置批量插入的记录数，减少BQ API请求次数，提升写入效率。

4. 优化上游计算流程 #

• 多表关联时，对小表使用broadcast函数，避免大表shuffle：

  import org.apache.spark.sql.functions.broadcast
  val joinedDs = bigTable.join(broadcast(smallTable), "join_key")
  

• 尽量减少map操作中的复杂逻辑，或者将部分逻辑转为Spark SQL内置函数（内置函数经过优化，比自定义map更高效）。

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