小菜鸟

java菜鸟号正在起航

0%

spark连接jdbc

发表于 更新于 分类于 阅读次数： 33 Valine： 0
本文字数： 5.1k 阅读时长 ≈ 5 分钟

spark连接jdbc数据库全指南：从 JdbcRDD 到 DataFrame

Spark 提供了多种方式连接关系型数据库（如 MySQL、PostgreSQL），实现数据的读写操作。其中，JdbcRDD 是早期基于 RDD 的数据库交互方式，而 DataFrame API 则提供了更简洁、高效的操作接口。本文将详细介绍 Spark 连接 JDBC 数据库的两种核心方法，包括使用 JdbcRDD 进行低阶操作，以及通过 DataFrame 实现高阶数据处理，并提供最佳实践与优化技巧。

基于 RDD 的 JDBC 连接：JdbcRDD

JdbcRDD 是 Spark 早期为 RDD 设计的 JDBC 交互工具，允许通过 SQL 查询从数据库读取数据并转换为 RDD。其核心思想是并行化执行 SQL 查询，通过分区键将数据分片到多个 Task 中读取。

JdbcRDD 的基本用法

核心构造参数

JdbcRDD 的构造函数需指定以下关键参数：

1
2
3
4
5
6
7
8
9
JdbcRDD(  
  sc: SparkContext,                // SparkContext 实例  
  getConnection: () => Connection, // 生成数据库连接的函数  
  sql: String,                     // 查询 SQL（需包含分区占位符）  
  lowerBound: Long,                // 分区键下界  
  upperBound: Long,                // 分区键上界  
  numPartitions: Int,              // 分区数  
  mapRow: (ResultSet) => T         // 结果集转换函数（将 ResultSet 映射为 RDD 元素）  
)
完整示例：读取 MySQL 数据
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
import org.apache.spark.rdd.JdbcRDD  
import java.sql.{Connection, DriverManager, ResultSet}  

object JdbcRDDExample {  
  def main(args: Array[String]): Unit = {  
    val sc = new SparkContext("local[*]", "JdbcRDDExample")  

    // 1. 定义数据库连接函数（每个分区创建一个连接）  
    def getConnection(): Connection = {  
      Class.forName("com.mysql.cj.jdbc.Driver") // MySQL 8.0+ 驱动  
      DriverManager.getConnection(  
        "jdbc:mysql://localhost:3306/test?useSSL=false&serverTimezone=UTC",  
        "root",   // 数据库用户名  
        "password" // 数据库密码  
      )  
    }  

    // 2. 定义结果集映射函数（将 ResultSet 转换为元组）  
    def mapRow(rs: ResultSet): (Int, String, Int) = {  
      (rs.getInt("id"), rs.getString("name"), rs.getInt("age"))  
    }  

    // 3. 创建 JdbcRDD（按 id 分区查询）  
    val jdbcRDD = new JdbcRDD(  
      sc,  
      getConnection,  
      sql = "SELECT id, name, age FROM user WHERE id >= ? AND id <= ?", // 带分区占位符的 SQL  
      lowerBound = 1,       // id 下界  
      upperBound = 100,     // id 上界  
      numPartitions = 3,    // 分为 3 个分区  
      mapRow = mapRow       // 结果映射函数  
    )  

    // 4. 处理数据  
    jdbcRDD.collect().foreach(println)  

    sc.stop()  
  }  
}

JdbcRDD 的核心特性与限制

优势
  • 并行读取：通过分区键（如示例中的 id）将查询分片，多个 Task 并行读取，提升效率；
  • 低阶控制：可自定义连接创建和结果映射逻辑，灵活性高。
限制
  • 仅支持读取：无法通过 JdbcRDD 写入数据到数据库；
  • 需手动管理分区：需指定分区键、上下界和分区数，无自动分区逻辑；
  • 代码繁琐：需手动处理数据库连接、结果集映射，易出错。

基于 DataFrame 的 JDBC 连接（推荐）

Spark SQL 提供了更高阶的 DataFrame API 用于 JDBC 交互，支持读写双向操作，且语法更简洁，性能更优。

从 JDBC 读取数据到 DataFrame

基本用法

通过 spark.read.jdbc 方法读取数据库表：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
import org.apache.spark.sql.SparkSession  

val spark = SparkSession.builder()  
  .appName("Spark JDBC Example")  
  .master("local[*]")  
  .getOrCreate()  

// 数据库连接参数  
val jdbcUrl = "jdbc:mysql://localhost:3306/test?useSSL=false&serverTimezone=UTC"  
val connectionProperties = new java.util.Properties()  
connectionProperties.put("user", "root")  
connectionProperties.put("password", "password")  
connectionProperties.put("driver", "com.mysql.cj.jdbc.Driver")  

// 读取整张表  
val df = spark.read.jdbc(  
  jdbcUrl,  
  table = "user",       // 表名或 SQL 子查询（需用括号包裹）  
  connectionProperties  
)  

df.show()  // 展示数据  
df.printSchema()  // 打印表结构
并行读取优化

与 JdbcRDD 类似，DataFrame 也支持按列分区并行读取，避免单 Task 读取全表：

1
2
3
4
5
6
7
8
9
10
// 按 id 列分区，分为 3 个分区  
val df = spark.read.jdbc(  
  jdbcUrl,  
  table = "user",  
  columnName = "id",    // 分区列（需为数值型）  
  lowerBound = 1L,      // 下界  
  upperBound = 100L,    // 上界  
  numPartitions = 3,    // 分区数  
  connectionProperties  
)
通过 SQL 查询读取

若需过滤或聚合数据，可直接传入 SQL 子查询：

1
2
3
4
5
6
// 读取查询结果（注意子查询需用括号包裹并命名）  
val df = spark.read.jdbc(  
  jdbcUrl,  
  table = "(SELECT id, name FROM user WHERE age > 18) AS subquery",  
  connectionProperties  
)

从 DataFrame 写入数据到 JDBC

通过 df.write.jdbc 方法将 DataFrame 数据写入数据库，支持多种写入模式（如追加、覆盖）。

基本用法
1
2
3
4
5
6
7
8
9
10
// 准备待写入的数据（示例：新增一条用户记录）  
import spark.implicits._  
val newData = Seq((101, "Alice", 25), (102, "Bob", 30)).toDF("id", "name", "age")  

// 写入数据库  
newData.write.jdbc(  
  jdbcUrl,  
  table = "user",  
  connectionProperties  
)
写入模式配置

通过 mode 方法指定写入策略：

1
2
3
4
5
6
newData.write  
  .mode("append")       // 追加模式（默认）  
  // .mode("overwrite")  // 覆盖模式（先删表再写入）  
  // .mode("ignore")    // 忽略模式（表存在则不写入）  
  // .mode("error")     // 错误模式（表存在则抛异常）  
  .jdbc(jdbcUrl, "user", connectionProperties)
批量写入优化

通过配置参数提升写入性能：

1
2
3
4
5
6
connectionProperties.put("batchsize", "1000")  // 批量写入大小（默认 1000）  
connectionProperties.put("isolationLevel", "NONE")  // 关闭事务隔离（提升性能）  

newData.write  
  .option("truncate", "true")  // 覆盖模式下使用 TRUNCATE 而非 DROP（保留表结构）  
  .jdbc(jdbcUrl, "user", connectionProperties)

JDBC 连接的核心配置参数

无论是 JdbcRDD 还是 DataFrame，都需配置数据库连接参数，以下是常用参数说明：

参数名 含义 示例值
url JDBC 连接地址 jdbc:mysql://localhost:3306/test
driver JDBC 驱动类名 MySQL 8.0+：com.mysql.cj.jdbc.Driver
user 数据库用户名 root
password 数据库密码 password
dbtable/table 表名或子查询 "user" 或 "(SELECT * FROM user) t"
partitionColumn 分区列（并行读取时） "id"
lowerBound 分区列下界 1L
upperBound 分区列上界 100L
numPartitions 分区数 3

最佳实践与性能优化

驱动包管理

  • 添加驱动依赖：需将数据库 JDBC 驱动包添加到 Spark 依赖中，可通过—jars命令行参数或pom.xml配置：

    1
    spark-submit --jars mysql-connector-java-8.0.28.jar your_app.jar

并行读取优化

  • 选择合适的分区列：优先使用自增主键或均匀分布的列作为分区列，避免数据倾斜；
  • 合理设置分区数：分区数建议与数据库连接池大小匹配，避免连接数过多导致数据库压力过大。

写入性能优化

  • 批量写入：通过 batchsize 参数设置批量提交大小（建议 1000~10000）；
  • 关闭事务：非关键场景下设置 isolationLevel=NONE 提升写入速度；
  • 避免小文件：写入前通过 repartition 调整分区数，减少数据库连接次数。

资源控制

  • 限制连接数：通过 spark.sql.execution.arrow.maxRecordsPerBatch 控制内存占用；

  • 超时设置：添加连接超时参数避免长时间阻塞：

    1
    jdbcUrl = "jdbc:mysql://localhost/test?connectTimeout=10000&socketTimeout=30000"

常见问题与解决方案

1. 驱动类找不到（ClassNotFoundException）

  • 原因：未添加 JDBC 驱动包或驱动类名错误；
  • 解决：确认驱动包版本与数据库匹配（如 MySQL 5.x 用 com.mysql.jdbc.Driver，8.x 用 com.mysql.cj.jdbc.Driver）。

2. 数据倾斜（部分 Task 耗时过长）

  • 原因：分区列数据分布不均，导致部分分区读取数据量过大；
  • 解决：更换分区列或手动拆分热点分区。

3. 写入权限不足

  • 原因：数据库用户无 INSERT 或 CREATE 权限；
  • 解决：为用户授予目标表的读写权限，或使用 append 模式避免表结构修改。
相关文章

欢迎关注我的其它发布渠道

表情 | 预览
快来做第一个评论的人吧~
Powered By Valine
v1.3.10