Spark將計算結果寫入到Mysql中

今天主要來談談如何將Spark計算的結果寫入到Mysql或者其他的關系型數據庫里面。其實方式也很簡單，代碼如下：

package scala
 
import java.sql.{DriverManager, PreparedStatement, Connection}
import org.apache.spark.{SparkContext, SparkConf}
 
object RDDtoMysql {
 
  case class Blog(name: String, count: Int)
 
  def myFun(iterator: Iterator[(String, Int)]): Unit = {
    var conn: Connection = null
    var ps: PreparedStatement = null
    val sql = "insert into blog(name, count) values (?, ?)"
    try {
      conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/spark", 
　　　　"root", "123456")
      iterator.foreach(data => {
        ps = conn.prepareStatement(sql)
        ps.setString(1, data._1)
        ps.setInt(2, data._2)
        ps.executeUpdate()
      }
      )
    } catch {
      case e: Exception => println("Mysql Exception")
    } finally {
      if (ps != null) {
        ps.close()
      }
      if (conn != null) {
        conn.close()
      }
    }
  }
 
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("RDDToMysql").setMaster("local")
    val sc = new SparkContext(conf)
    val data = sc.parallelize(List(("www", 10), ("iteblog", 20), ("com", 30)))
    data.foreachPartition(myFun)
  }
}

其實是通過foreachPartition遍歷RDD的每個分區，並調用普通的Scala方法來寫數據庫。在運行程序之前需要確保數據庫里面存在blog表，可以通過下面語句創建：

CREATE TABLE `blog` (
  `name` varchar(255) NOT NULL,
  `count` int(10) unsigned DEFAULT NULL
) ENGINE=InnoDB DEFAULT CHARSET=utf-8

然后直接運行上述的代碼即可。運行完成你就可以在數據庫里面查詢結果：

SELECT * FROM blog b;
www　　10
iteblog　　20
com　　30

需要注意的是：
　　1、你最好使用foreachPartition 函數來遍歷RDD，並且在每台Work上面創建數據庫的connection。
　　2、如果你的數據庫並發受限，可以通過控制數據的分區來減少並發。
　　3、在插入Mysql的時候最好使用批量插入。
　　4、確保你寫入數據庫過程能夠處理失敗，因為你插入數據庫的過程可能會經過網絡，這可能會導致數據插入到數據庫失敗。
　　5、不建議將你的RDD數據寫入到Mysql等關系型數據庫中。