[Spark streaming举例]-- 实时统计并且存储到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()
  }
}