package SparkDemoimport java.sql.{Connection, DriverManager, PreparedStatement}import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}object DStreamToMySQL {//定义更新函数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))//设置日志等级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存入数据库//stateWordCount中包含一堆的Rdd//我们需要对每个Rdd中的每条数据进行处理储存stateWordCount.foreachRDD(rdd => {//每个rdd中包含的数据类型为(String,Int)//我们把所有数据records定义为Iterator类型，方便我们遍历def func(records:Iterator[(String,Int)]): Unit ={//注意，conn和stmt定义为var不能是valvar conn: Connection = nullvar stmt : PreparedStatement = nulltry{//连接数据库val url = "jdbc:mysql://localhost:3306/spark" //地址+数据库val user = "root"val password = ""conn = DriverManager.getConnection(url,user,password)//records.foreach(p =>{//wordcount为表名，word和count为要插入数据的属性//插入数据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重新分区repairtitionedRDD.foreachPartition(func)//对重新分区后的rdd执行func函数})ssc.start()//启动ssc.awaitTermination()//等待终止命令}}