package SparkDemoimport java.sql.{Connection, DriverManager, PreparedStatement}import org.apache.spark.SparkConf import org.apache.spark.streaming.{Seconds, StreamingContext}object DStreamToMySQL {//定義更新函數(shù)def updateFunc(newValues : Seq[Int],state :Option[Int]):Option[Int] = {val currentCount = newValues.foldLeft(0)(_+_)val previousCount = state.getOrElse(0)Some(currentCount+previousCount)}def main(args : Array[String]): Unit ={//建立SparkStreamval conf = new SparkConf().setAppName("DStreamToMySQL")val ssc = new StreamingContext(conf,Seconds(1))//設(shè)置日志等級StreamingLoggingExample.setStreamingLogLevels()val lines = ssc.textFileStream("/tmp/yuhang.zhang/data")val words = lines.flatMap(_.split(" "))val pairWord = words.map((_,1))//累計更新val stateWordCount = pairWord.updateStateByKey[Int](updateFunc)//將stateWordCount存入數(shù)據(jù)庫//stateWordCount中包含一堆的Rdd//我們需要對每個Rdd中的每條數(shù)據(jù)進(jìn)行處理儲存stateWordCount.foreachRDD(rdd => {//每個rdd中包含的數(shù)據(jù)類型為(String,Int)//我們把所有數(shù)據(jù)records定義為Iterator類型，方便我們遍歷def func(records:Iterator[(String,Int)]): Unit ={//注意，conn和stmt定義為var不能是valvar conn: Connection = nullvar stmt : PreparedStatement = nulltry{//連接數(shù)據(jù)庫val url = "jdbc:mysql://localhost:3306/spark" //地址+數(shù)據(jù)庫val user = "root"val password = ""conn = DriverManager.getConnection(url,user,password)//records.foreach(p =>{//wordcount為表名，word和count為要插入數(shù)據(jù)的屬性//插入數(shù)據(jù)val sql = "insert into wordcount(word,count) values(?,?)"stmt = conn.prepareStatement(sql)stmt.setString(1,p._1.trim)stmt.setInt(2,p._2.toInt)stmt.executeUpdate()})}catch {case e : Exception => e.printStackTrace()}finally {if(stmt != null)stmt.close()if(conn != null)conn.close()}}val repairtitionedRDD = rdd.repartition(3)//將每個rdd重新分區(qū)repairtitionedRDD.foreachPartition(func)//對重新分區(qū)后的rdd執(zhí)行func函數(shù)})ssc.start()//啟動ssc.awaitTermination()//等待終止命令}}

　　

轉(zhuǎn)載于:https://www.cnblogs.com/zzhangyuhang/p/9075992.html

總結(jié)

以上是生活随笔為你收集整理的【sparkStreaming】将DStream保存在MySQL的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。