sparkstreaming实时统计并且存储到mysql数据库中

package com.scala.my

import org.apache.spark.SparkConf

import org.apache.spark.streaming.Durations

import org.apache.spark.streaming.StreamingContext

/**

 * 

 * @author root

 * 测试步骤：

 *    1\打开h15\h16\h17\h18,启动zookeeper，再启动hadoop集群：start-all.sh，再启动mysql

 *    2\在h15上创建文件夹wordcount_checkpoint，用于docheckpoint

 *       在h5上mysql的dg数据库中创建表t_word

 *    3\启动eclipse的本程序，让他等待着

 *    4\在h15的dos窗口下输入单词,以空格分隔的单词(需要在h15上开启端口9999:#nc -lk 9999)

 *    5\查询h15上的mysql的dg数据库的t_word表是否有数据即可

 *

 * 注：建表语句

 *     mysql> show create table wordcount;  //查看表语句
CREATE TABLE   t_word (
id  int(11) NOT NULL AUTO_INCREMENT,
updated_time  timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
word varchar(255) DEFAULT NULL,
count  int(11) DEFAULT NULL,
PRIMARY KEY (id)
);

 */

 * 

 * 测试结果：通过，注意-----》第74行没有取得数据，原因在最后没有触发事件（封装事件），目前已经解决

 * 

 * sh spark-submit --master spark://de2:7077 --class 全类名 --driver-class-path /mysql-connector-java-5.1.26.jar  sparkstreaming.jar

   sh spark-submit --class com.day6.scala.my.PresistMysqlWordCount --master yarn-cluster --driver-class-path /home/spark-1.5.1-bin-hadoop2.4/lib/mysql-connector- 

    java-5.1.31-bin.jar /home/spark-1.5.1-bin-hadoop2.4/sparkstreaming.jar 

$bin/hadoop dfsadmin -safemode leave

 

也就是关闭Hadoop的安全模式，这样问题就解决了。

 */

object PresistMysqlWordCount {

  def main(args: Array[String]): Unit = {

    //获取streamingContext，并且设置每5秒切割一次rdd

//    val sc = new StreamingContext(new SparkConf().setAppName("mysqlPresist").setMaster("local[2]"), Durations.seconds(8))

    val sc = new StreamingContext(new SparkConf().setAppName("mysqlPresist").setMaster("local[2]"), Durations.seconds(8))

    //设置checkpoit缓存策略

    /**

     * 利用 checkpoint 来保留上一个窗口的状态，

     * 这样可以做到移动窗口的更新统计

     */

    sc.checkpoint("hdfs://hh15:8020/wordcount_checkpoint")

//    sc.checkpoint("hdfs://h15:8020/wordcount_checkpoint")

    //获取doc窗口或者hdfs上的words

    //    val lines=sc.textFileStream("hdfs://h15:8020/文件夹名称")  //实时监控hdfs文件夹下新增的数据

    val lines = sc.socketTextStream("hh15", 9999)

//    val lines = sc.socketTextStream("h15", 9999)

    //压扁

    val words = lines.flatMap { x => x.split(" ") }

    //map

    val paris = words.map { (_, 1) }

    //定义一个函数,用于保持状态

    val addFunc = (currValues: Seq[Int], prevValueState: Option[Int]) => {

      var newValue = prevValueState.getOrElse(0)

      for (value <- currValues) {

        newValue += value

      }

      //返回option

      Option(newValue)

      //        //通过Spark内部的reduceByKey按key规约，然后这里传入某key当前批次的Seq/List,再计算当前批次的总和

      //        val currentCount = currValues.sum

      //        // 已累加的值

      //        val previousCount = prevValueState.getOrElse(0)

      //        // 返回累加后的结果，是一个Option[Int]类型

      //        Some(currentCount + previousCount)

    }

    //updateStateByKey操作

    /**

     * updateStateByKey 解释:

     * 以DStream中的数据进行按key做reduce操作，然后对各个批次的数据进行累加

     * 在有新的数据信息进入或更新时，可以让用户保持想要的任何状。使用这个功能需要完成两步：

     * 1) 定义状态：可以是任意数据类型

     * 2) 定义状态更新函数：用一个函数指定如何使用先前的状态，从输入流中的新值更新状态。

     * 对于有状态操作，要不断的把当前和历史的时间切片的RDD累加计算，随着时间的流失，计算的数据规模会变得越来越大。

     */

    val end = paris.updateStateByKey(addFunc)

    //插入mysql数据库

   end.foreachRDD(wd => wd.foreachPartition(

      data => {

        val conn = ConnectPool.getConn("root", "1714004716", "hh15", "dg")

//        val conn = ConnectPool.getConn("root", "1714004716", "h15", "dg")

        //插入数据

//        conn.prepareStatement("insert into t_word2(word,num) values('tom',23)").executeUpdate()

        try {

          for (row <- data) {

            println("input data is " + row._1 + "  " + row._2)

            val sql = "insert into t_word2(word,num) values(" + "'" + row._1 + "'," + row._2 + ")"

            conn.prepareStatement(sql).executeUpdate()

          }

        }finally {

          conn.close()

        }

      }))

      

    //必须添加end.print(),触发封装事件

     end.print()

    //开启接收模式

    sc.start()

    //等待

    sc.awaitTermination()

    //关闭资源

    sc.stop()

  }

}