在Databricks中使用JDBC作为流数据源连接时遇类找不到错误求助

核心问题原因 #

1. 版本不兼容：你的集群使用Spark 3.3.0，但安装的spark-streaming_2.12:3.3.2与集群内置Spark版本存在差异，且第三方库sutugin/spark-streaming-jdbc-source未适配Spark 3.3.x及Databricks定制环境。
2. API变更：StreamWriteSupport属于Spark SQL V2数据源旧版接口，在Spark 3.3.x或Databricks的定制Spark发行版中，该类已被移除、重命名或调整，导致依赖该类的第三方库无法加载。
3. 库维护停滞：该第三方仓库长期未更新，未跟进Spark 3.x的API迭代，无法适配当前集群环境。

替代解决方案 #

在Databricks中实现JDBC流读取，推荐使用结构化流+增量轮询的方式，无需依赖第三方库，具体步骤如下：

1. 基于自增ID/时间戳的增量读取 #

利用数据库中的自增主键或时间戳字段，定期轮询读取新增数据：

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
import org.apache.spark.sql.streaming.Trigger

val spark = SparkSession.builder().getOrCreate()

// 初始化checkpoint路径，用于记录上次读取的最大ID
val checkpointPath = "/dbfs/path/to/checkpoint"
var lastMaxId = 0L

// 读取checkpoint中记录的上次最大ID
if (dbutils.fs.exists(checkpointPath)) {
  lastMaxId = spark.read.text(checkpointPath).head().getString(0).toLong
}

// 定义增量读取函数
def readIncrementalData(): Unit = {
  val df = spark.read.format("jdbc")
    .option("url", "jdbc:postgresql://your-host:5432/db-name")
    .option("dbtable", s"(SELECT * FROM your_table WHERE id > $lastMaxId) AS incremental_data")
    .option("user", "username")
    .option("password", "password")
    .option("driver", "org.postgresql.Driver")
    .load()

  if (!df.isEmpty) {
    // 处理数据（示例：写入Delta表）
    df.write.mode("append").saveAsTable("your_delta_table")
    
    // 更新checkpoint中的最大ID
    val currentMaxId = df.agg(max("id")).head().getLong(0)
    spark.sparkContext.parallelize(Seq(currentMaxId.toString)).saveAsTextFile(checkpointPath)
    lastMaxId = currentMaxId
  }
}

// 设置定时触发，每30秒执行一次增量读取
val query = spark.readStream
  .format("rate")
  .option("rowsPerSecond", 1)
  .load()
  .writeStream
  .foreachBatch { (_, _) =>
    readIncrementalData()
  }
  .trigger(Trigger.ProcessingTime("30 seconds"))
  .start()

query.awaitTermination()

2. 使用Databricks JDBC批量读取优化 #

如果需要更高效的批量处理，可结合Databricks的foreachBatch与JDBC批量读取参数提升性能：

// 在readIncrementalData函数中添加批量参数
val df = spark.read.format("jdbc")
  .option("url", "jdbc:postgresql://your-host:5432/db-name")
  .option("dbtable", s"(SELECT * FROM your_table WHERE id > $lastMaxId) AS incremental_data")
  .option("user", "username")
  .option("password", "password")
  .option("driver", "org.postgresql.Driver")
  .option("fetchsize", "10000") // 批量读取大小
  .load()

关键注意事项 #

• 确保已安装PostgreSQL JDBC驱动（org.postgresql:postgresql:42.5.4或兼容版本），可通过Databricks集群库管理界面安装。
• 若使用时间戳字段作为增量依据，需确保数据库中该字段有索引，避免全表扫描。
• Checkpoint路径建议使用DBFS路径，确保集群重启后能保留上次读取的位置。

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