spark - 将RDD保存到RMDB(MYSQL)数据库中

92 阅读 0 评论 61 点赞

我是靠谱客的博主英勇指甲油，这篇文章主要介绍spark - 将RDD保存到RMDB(MYSQL)数据库中，现在分享给大家，希望可以做个参考。

SCALA连接数据库批量插入：

scala> import java.sql.DriverManager

scala> var url = "jdbc:mysql://localhost:3306/mydb?useUnicode=true&characterEncoding=utf8"

scala> var username = "cui"

scala> var password = "dbtest"

scala> val conn= DriverManager.getConnection(url,username,password)

scala> val pstat = conn.prepareStatement ("INSERT INTO `TEST` (`ID`, `AGE`) VALUES (?, ?)")

scala> pstat.clearBatch

scala> pstat.setInt(1,501)

scala> pstat.setInt(2,501)

scala> pstat.addBatch

scala> pstat.setInt(1,502)

scala> pstat.setInt(2,502)

scala> pstat.addBatch

scala> pstat.setInt(1,503)

scala> pstat.setInt(2,503)

scala> pstat.addBatch

scala> pstat.executeBatch
res24: Array[Int] = Array(1, 1, 1)

RDD保存到数据库：

Just use foreachPartition to create and execute a SQL statement via JDBC over a batch of records. The code is just normal JDBC code.

https://community.cloudera.com/t5/Advanced-Analytics-Apache-Spark/Spark-Streaming-save-output-to-mysql-DB/td-p/25607

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
import java.sql.DriverManager

var data = sc.parallelize(Array( (1,10) ,(1,100), (1,1000), (1,10000),(2,10) ,(2,100), (2,1000), (2,10000)  ),2 )

data.foreachPartition(
    it =>{
        var url = "jdbc:mysql://localhost:3306/mydb?useUnicode=true&characterEncoding=utf8"
        val conn= DriverManager.getConnection(url,"username","password")
        val pstat = conn.prepareStatement ("INSERT INTO `TEST` (`ID`, `AGE`) VALUES (?, ?)")
        for (obj <-it){
            pstat.setInt(1,obj._1)
            pstat.setInt(2,obj._2)
            pstat.addBatch
        }
        try{
            pstat.executeBatch
        }finally{
            pstat.close
            conn.close
        }
    }
)