spark中rdd保存到mysql_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.

importjava.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

pstat.setInt(1,obj._1)

pstat.setInt(2,obj._2)

pstat.addBatch

}try{

pstat.executeBatch

}finally{

pstat.close

conn.close

}

}

)