最大化Spark JDBC读取吞吐量：如何拆分任务实现JDBC读取与Hive写入分离？

首先得给你点破原代码的核心问题：你用Driver端的ForkJoinPool并行提交Spark Job，这种方式其实没利用好Spark的分布式特性，反而容易踩JDBC连接的坑——每个Job自己又占100个cores，总连接数直接爆炸；而且读写绑在同一个任务里，写Hive的时间会占着JDBC连接，导致连接没法一直用来读，利用率上不去。

想要实现“100个读任务持续拉取JDBC，100个写任务持续写入Hive”的目标，最靠谱的是用Spark原生的分布式并行能力，下面给你两个实用方案：

方案一：极简流水线式读写（最推荐） #

这个方案直接利用Spark的Stage流水线特性，读和写自动并行，刚好匹配JDBC的100连接上限：

步骤1：精准控制JDBC并行读取 #

放弃Driver端的ForkJoinPool，用Spark JDBC数据源自带的numPartitions参数，直接把并行读取的任务数设为100，刚好卡满JDBC的连接上限。需要注意的是，得指定一个用来拆分数据的列（比如ID、时间戳这类有序列），Spark会自动把数据分成100个分区，每个分区对应一个JDBC连接：

val jdbcDF = spark.read
  .format("jdbc")
  .option("url", "你的JDBC地址")
  .option("dbtable", "目标表名")
  .option("user", "用户名")
  .option("password", "密码")
  .option("numPartitions", 100) // 核心：强制100个并行读任务
  .option("partitionColumn", "用来拆分的列（比如id）") // 必须是数值/时间类型
  .option("lowerBound", "列的最小值")
  .option("upperBound", "列的最大值")
  .option("fetchSize", "10000") // 可选优化：每次拉取更多数据，减少IO次数
  .load()

步骤2：并行写入Hive #

读取到的DataFrame已经是100个分区了，直接写入Hive时，Spark会自动启动100个并行写任务，而且是流水线式执行——读任务拉完一个分区的数据，写任务立刻开始写入这个分区，读任务继续拉下一个分区。这样就实现了“100读+100写”同时跑，完全没有中间等待，利用率拉满：

jdbcDF.write
  .mode("append")
  .partitionBy("somecolumn") // 按你的业务列分Hive分区
  .format("hive")
  .saveAsTable("你的Hive表名")

方案二：完全分离读写任务（适合复杂业务场景） #

如果你的业务需要在读写之间加复杂处理，或者想把读写拆成完全独立的任务流，可以用Spark结构化流做中间缓冲：

步骤1：启动Hive写入的消费者流 #

先创建一个内存流作为数据缓冲，然后启动流查询，专门负责把数据写入Hive：

import org.apache.spark.sql.streaming.Trigger

// 先定义好和JDBC数据一致的Schema（可以先读一次JDBC获取）
val dataSchema = spark.read
  .format("jdbc")
  .option("url", "你的JDBC地址")
  .option("dbtable", "目标表名")
  .option("user", "用户名")
  .option("password", "密码")
  .load()
  .schema

// 创建内存流作为数据源
val streamReader = spark.readStream
  .schema(dataSchema)
  .format("memory")
  .load("jdbc_data_stream")

// 启动流写入Hive，设置并行度匹配读任务
val writeQuery = streamReader.writeStream
  .foreachBatch { (batchDF, _) =>
    batchDF.write
      .mode("append")
      .partitionBy("somecolumn")
      .format("hive")
      .saveAsTable("你的Hive表名")
  }
  .trigger(Trigger.ProcessingTime("0 seconds")) // 有数据就立即处理
  .option("checkpointLocation", "/tmp/hive_write_checkpoint") // 必须设置 checkpoint 路径
  .start()

步骤2：启动JDBC读取的生产者任务 #

用100个并行任务读取JDBC数据，然后写入内存流，这样读任务是生产者，写任务是消费者，两者并行运行：

val jdbcDF = spark.read
  .format("jdbc")
  .option("url", "你的JDBC地址")
  .option("dbtable", "目标表名")
  .option("user", "用户名")
  .option("password", "密码")
  .option("numPartitions", 100)
  .option("partitionColumn", "用来拆分的列")
  .option("lowerBound", "最小值")
  .option("upperBound", "最大值")
  .load()

// 写入内存流，触发消费端的写入任务
jdbcDF.write
  .format("memory")
  .mode("append")
  .save("jdbc_data_stream")

// 等待所有写入任务完成
writeQuery.awaitTermination()

额外优化小贴士 #

• 确保Executor资源刚好匹配：M*N=100，让每个读/写任务都有独立的core，避免任务排队拖慢速度。
• 关闭自动广播：如果JDBC数据量很大，设置spark.sql.autoBroadcastJoinThreshold=-1，避免Driver端内存溢出。
• 监控JDBC连接：可以在数据源端查看连接数，确保稳定在100左右，避免出现连接泄漏。

备注：内容来源于stack exchange，提问作者Alexey