如何将spark streaming處理結果儲存到關系型資料庫中

spark streaming是一個分布式高可靠的準實時處理系統，其資料源可以flume、Hdfs、kafka等，其結果可以儲存到關系型資料庫，HDFS上。儲存到HDFS上相對簡單，一句話就可以搞定，但是要儲存到關系資料庫中，相對比較麻煩，既要連結資料庫，又要知道資料字段。

我們首先寫個wordcount程式測試一下，通過網絡發資料到spark streaming

發資料程式如下

import java.io.{PrintWriter}

import java.net.ServerSocket
import scala.io.Source


object SaleSimulation {
  
 def index(length: Int) = {

    import java.util.Random
 
   val rdm = new Random

    rdm.nextInt(length)

  }

  
def main(args: Array[String]) {
   
 if (args.length != 3) {
  
    System.err.println("Usage: <filename> <port> <millisecond>")
 
     System.exit(1)
  
  }

  
  val filename = args(0)
 
   val lines = Source.fromFile(filename).getLines.toList

    val filerow = lines.length

       val listener = new ServerSocket(args(1).toInt)
    
while (true) {
 
     val socket = listener.accept()

      new Thread() {
   
     override def run = {
   
       println("Got client connected from: " + socket.getInetAddress)
   
       val out = new PrintWriter(socket.getOutputStream(), true)
  
        while (true) {
    
        Thread.sleep(args(2).toLong)
  
         val content = lines(index(filerow))
    
         println(content)
      
         out.write(content + '\n')
  
          out.flush()
    
      }
   
       socket.close()
   
     }
      }.start()
  
  }
  
}

}
           

打成jar包後運作

java -cp spark_streaming_test.jar com.pinganfu.ss.SaleSimulation /spark/people.txt 9999 1000

           

spark streaming程式如下：

import java.sql.{PreparedStatement, Connection, DriverManager}  
import java.util.concurrent.atomic.AtomicInteger  
import org.apache.spark.SparkConf  
import org.apache.spark.streaming.{Seconds, StreamingContext}  
import org.apache.spark.streaming._  
import org.apache.spark.streaming.StreamingContext._  
//No need to call Class.forName("com.mysql.jdbc.Driver") to register Driver?  
object SparkStreamingForPartition {  
  def main(args: Array[String]) {  
    val conf = new SparkConf().setAppName("NetCatWordCount")  
    conf.setMaster("local[3]")  
    val ssc = new StreamingContext(conf, Seconds(5))   
    val dstream = ssc.socketTextStream("hadoopMaster", 9999).flatMap(_.split(" ")).map(x => (x, 1)).reduceByKey(_ + _)  
    dstream.foreachRDD(rdd => {  
      //embedded function  
      def func(records: Iterator[(String,Int)]) {  
//Connect the mysql
        var conn: Connection = null  
        var stmt: PreparedStatement = null  
        try {  
          val url = "jdbc:mysql://hadoopMaster:3306/streaming";  
          val user = "root";  
          val password = "hadoop"  
          conn = DriverManager.getConnection(url, user, password)  
          records.foreach(word => {  
            val sql = "insert into wordcounts values (?,?)";  
            stmt = conn.prepareStatement(sql);  
            stmt.setString(1, word._1)  
            stmt.setInt(2, word._2)  
            stmt.executeUpdate();  
          })  
        } catch {  
          case e: Exception => e.printStackTrace()  
        } finally {  
          if (stmt != null) {  
            stmt.close()  
          }  
          if (conn != null) {  
            conn.close()  
          }  
        }  
      }   
      val repartitionedRDD = rdd.repartition(3)  
      repartitionedRDD.foreachPartition(func)  
    })  
    ssc.start()  
    ssc.awaitTermination()  
  }  
}  
           

運作結果

1. DStream.foreachRDD是一個Output Operation,DStream.foreachRDD是資料落地很常用的方法

2. 擷取MySQL Connection的操作應該放在foreachRDD的參數（是一個RDD[T]=>Unit的函數類型)，這樣，當

foreachRDD方法在每個Worker上執行時，連接配接是在Worker上建立。如果Connection的擷取放到dstream.foreachRDD之

前，那麼Connection的擷取動作将發生在Driver端，然後通過序列化的方式發送到各個Worker(Connection的序列化通常是無法正确序列化的）

3. Connection的擷取在foreachRDD的參數中擷取，同時還要在周遊RDD之前擷取（調用RDD的foreach方法前擷取)，如果周遊中擷取，那麼RDD中的每個record都要打開關閉連接配接，這對于資料庫連接配接資源将是極大的考驗

4. 業務邏輯處理定義在func中，它是在foreachRDD的方法參數體中定義的，如果把func的定義放到外面，即Driver中，貌似也是可以的，Spark會對計算方法通過Broadcast進行廣播到各個計算節點。