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

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

        }

    }

)