为何不存在JDBC Spark Streaming Receiver？批量处理大表方案咨询

你提到的分批读取大型JDBC表、用Spark Streaming处理且无需监控新数据的思路，其实是有合理性的，但确实很少有现成的JDBC Receiver实现，背后主要有这几个原因：

• Spark Streaming的定位偏向流式数据，而非批量回溯：Spark Streaming从设计初衷来看，是为了处理持续产生的实时流式数据（比如Kafka、Flume这类源源不断产生数据的数据源）。而你需求里的"一次性分批读取全表"，本质上更偏向离线批量处理场景。社区里更多会直接用Spark Core或者Spark SQL的JDBC数据源配合分页逻辑来实现，而非套上Spark Streaming的Receiver模式。

• Receiver的设计逻辑和你的需求不匹配：Receiver的核心是要持续监听、拉取新数据，保持长期运行状态。但你的需求是一次性读取完整个表就结束，不需要后续监控新行。这种情况下，Receiver的"持续运行"特性反而显得多余——你需要的只是一个能分批加载数据的任务，而非长期驻留的流接收器。

• Stratio的实现逻辑符合批量场景特性：你提到的Stratio/datasource-receiver会先把数据全读进DataFrame再处理，其实也是因为对于一次性全表读取的场景来说，用Spark SQL的JDBC数据源做分页加载，再转换成DStream的成本更低，完全没必要单独开发一个Receiver来做这件事。

如果你确实想用Spark Streaming实现，这里给一个简易的自定义Receiver思路 #

你可以自己写一个轻量的自定义Receiver，核心就是在Receiver启动后，按分页逻辑分批拉取JDBC表数据，每拉一批就推送到DStream，全部拉完后自动停止。

给你一段伪代码参考：

import org.apache.spark.streaming.receiver.Receiver
import org.apache.spark.storage.StorageLevel
import java.sql.{DriverManager, ResultSet}

class JDBCBatchReceiver(jdbcUrl: String, tableName: String, batchSize: Int) 
  extends Receiver[String](StorageLevel.MEMORY_AND_DISK_2) {

  override def onStart(): Unit = {
    // 启动一个单独线程处理数据拉取
    new Thread("JDBC Batch Fetch Thread") {
      override def run(): Unit = fetchBatchData()
    }.start()
  }

  private def fetchBatchData(): Unit = {
    var connection = null
    try {
      connection = DriverManager.getConnection(jdbcUrl)
      var offset = 0
      var hasMoreData = true

      while (hasMoreData && !isStopped()) {
        val query = s"SELECT * FROM $tableName LIMIT $batchSize OFFSET $offset"
        val stmt = connection.prepareStatement(query)
        val rs = stmt.executeQuery()

        val batchData = collection.mutable.ArrayBuffer[String]()
        var rowCount = 0
        while (rs.next()) {
          // 这里根据你的需求把行数据转换成字符串（或者其他格式）
          val rowStr = rs.getString("id") + "," + rs.getString("content")
          batchData.append(rowStr)
          rowCount += 1
        }

        if (rowCount == 0) {
          hasMoreData = false
        } else {
          // 将当前批次数据推送到DStream
          store(batchData.toArray)
          offset += batchSize
        }

        rs.close()
        stmt.close()
      }
      // 数据全部拉取完成后停止Receiver
      stop("All batch data fetched successfully")
    } catch {
      case e: Exception =>
        restart(s"Error fetching JDBC data: ${e.getMessage}", e)
    } finally {
      if (connection != null) connection.close()
    }
  }

  override def onStop(): Unit = {
    // 停止时清理资源，比如关闭JDBC连接（上面finally已经处理，这里可以补充其他逻辑）
  }
}

使用这个Receiver的示例：

import org.apache.spark.streaming.{StreamingContext, Seconds}
import org.apache.spark.SparkConf

val conf = new SparkConf().setAppName("JDBCBatchStreaming")
val ssc = new StreamingContext(conf, Seconds(5)) // 批次间隔这里可以随便设，因为我们是一次性拉取

val jdbcStream = ssc.receiverStream(
  new JDBCBatchReceiver("jdbc:mysql://your-host:3306/your-db", "large_table", 10000)
)

// 后续处理逻辑，比如打印或者写入存储
jdbcStream.foreachRDD { rdd =>
  println(s"Processing batch with ${rdd.count()} records")
  // 这里写你的业务处理代码
}

ssc.start()
ssc.awaitTermination()

更推荐的替代方案：用Spark SQL分页批量处理 #

其实对于你这种"一次性读取全表"的需求，用Spark SQL的JDBC数据源做分区读取会更高效，完全不需要依赖Spark Streaming：

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder().appName("JDBCBatchRead").getOrCreate()

val connectionProps = new java.util.Properties()
connectionProps.setProperty("user", "your-user")
connectionProps.setProperty("password", "your-pass")

// 按主键id分区间读取，自动分批加载
val df = spark.read.jdbc(
  url = "jdbc:mysql://your-host:3306/your-db",
  table = "large_table",
  columnName = "id", // 用来分区间的字段，最好是主键或有索引的字段
  lowerBound = 1,
  upperBound = 1000000, // 这个字段的最大值
  numPartitions = 100, // 分成多少个分区（也就是多少批）
  connectionProperties = connectionProps
)

// 处理每个分区的数据
df.foreachPartition { partition =>
  // 这里写你的业务逻辑，每个分区对应一批数据
}

这种方式利用Spark的分布式能力自动分批加载数据，性能和维护性都比自定义Receiver更好，更适合你的需求。

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