Spark Streaming 进阶实战五个例子

「爱情、让人受尽委屈。」 2023-03-01 08:59 55阅读 0赞

参考文章：[Spark Streaming 进阶实战五个例子][Spark Streaming]

[Spark Streaming进阶][Spark Streaming 1]

## 一、带状态的算子：UpdateStateByKey ##

实现 计算 过去一段时间到当前时间 单词 出现的 频次

object StatefulWordCount {
      def main(args: Array[String]): Unit = {
        val sparkConf = new SparkConf().setMaster("local[2]").setAppName("StatefulWordCount")
        val ssc = new StreamingContext(sparkConf, Seconds(5))
        //如果使用了 stateful  的算子，必须要设置 checkpoint，
        //因为老的值 必须要 存在 某个 目录下面，新的值 才能去更新老的值
        //在生产环境中，建议把checkpoint 设置到 hdfs 的某个文件夹中
        ssc.checkpoint(".")
        val lines = ssc.socketTextStream("localhost", 6789)
        val result = lines.flatMap(_.split(" ").map((_, 1)))
        val state  =  result.updateStateByKey[Int](updateFunction _)
        state.print()
        ssc.start()
        ssc.awaitTermination()
      }
     
      /**
        * 把当前的数据去更新已有的或者老的数据
        * @param currentValues  当前的
        * @param preValues  老的
        * @return
        */
      def updateFunction(currentValues: Seq[Int], preValues: Option[Int]): Option[Int] = {
        val current = currentValues.sum
        val pre = preValues.getOrElse(0)
        Some(current + pre)
      }
    }

## 二、实战：计算到目前为止累积出现的单词的个数写入到mysql中 ##

*  使用Spark Streaming进行词频统计
 *  Spark Streaming统计结果写入到MySQL中去

/**
      * 使用 spark  streaming  完成 词频统计，并输出到 mysql 数据库
      * 创建 数据库
      *
      * 创建数据表
      * create  table  wordcount (
      * word  varchar(50) default null,
      * wordcount int(10) default null
      * )
      */
    object ForeachRDDApp {
      def main(args: Array[String]): Unit = {
        val sparkConf = new SparkConf().setMaster("local[2]").setAppName("ForeachRDDApp")
        val ssc = new StreamingContext(sparkConf, Seconds(5))
      
        // 接收来自端口为6789的socket服务的消息
        // 命令行启动socket: nc -lk 6789
        val lines = ssc.socketTextStream("localhost", 6789)
        val result = lines.flatMap(_.split(" ").map((_, 1))).reduceByKey(_ + _)
        //result.print()
     
        //将结果写入到mysql
        //1、错误的方式
        //    result.foreachRDD(rdd =>{
        //        val connection =  createConnection()
        //        rdd.foreach {
        //           record =>
        //           val sql = "insert into wordcount (word,wordcount)" +
        //             "values ('"+record._1+"','"+record._2+"')"
        //          connection.createStatement().execute(sql)
        //        }
        //    })
     
        //2、正确的方式
        result.foreachRDD(rdd => {
          rdd.foreachPartition(partitionOfRecords => {
            if (partitionOfRecords.size > 0) {
              val connection = createConnection()
              partitionOfRecords.foreach(pair => {
                val sql = "insert into wordcount (word,wordcount)" +
                  "values ('" + pair._1 + "','" + pair._2 + "')"
                connection.createStatement().execute(sql)
              })
              connection.close()
            }
          })
        })
     
        //3、更好的方式，查阅官方文档，使用 连接池的方式
     
        //存在的问题，这样每次都会插入新的数据，同样的单词频次字段不会去累加更新
        //解决方案 ：每次 insert 之前，判断一下，该单词是否已经存在数据库中，如果已经存在则update
        //或者 存放在 hbase /redis 中，调用相应的api ，直接 插入和更新。
        ssc.start()
        ssc.awaitTermination()
      }
     
      /**
      * 获取MySQL的连接
      */
      def createConnection() = {
        Class.forName("com.mysql.jdbc.Driver")
        DriverManager.getConnection("jdbc://mysql://localhost:3306/dzx_spark", "root", "1234")
      }
    }

## 三、基于window 的统计 ##

>  *  window ：定时的进行一个时间段内的数据处理
>  *  window length ： 窗口的长度
>  *  sliding interval ： 窗口的间隔
>  *  这2个参数和我们的batch size  成倍数关系。如果不是倍数关系运行直接报错
>  *  每隔多久计算某个范围内的数据：每隔10秒计算前10分钟的wc ==>每隔 sliding interval  统计 window length 的值pair.reduceByKeyAndWindow((a:Int,b:Int)=>(a+b),Seconds(30),Seconds(10))

## 四、黑名单过滤 ##

*  transform 算子的使用
 *  Spark Streaming整合RDD进行操作

/**
      * 黑名单过滤
      *
      * 访问日志 ==> DStream
      *
      * 20180808,zs
      * 20180808,ls
      * 20180808,ww
      *
      * ==> (zs:20180808,zs) (ls:20180808,ls)(ww:20180808,ww)
      *
      * 黑名单列表 ==》 RDD
      * zs  ls
      * ==>(zs:true) (ls:true)
      * 
      * leftjoin
      * (zs:[<20180808,zs>,<true>])   pass ...
      * (ls:[<20180808,ls>,<true>])   pass ...
      * (ww:[<20180808,ww>,<false>]) ==> tuple1   ok...
      */
    object BlackNameListApp {
      def main(args: Array[String]): Unit = {
        val sparkConf = new SparkConf().setMaster("local[2]").setAppName("ForeachRDDApp")
        // 创建StreamingContext需要两个参数：SparkConf和batch interval
        val ssc = new StreamingContext(sparkConf, Seconds(5))
      
        /**
          * 构建黑名单
          */
        val blacks = List("zs", "ls")
        val blacksRDD = ssc.sparkContext.parallelize(blacks).map(x => (x, true))
        val lines = ssc.socketTextStream("localhost", 6789)
        val clicklog = lines
                         .map(x => (x.split(",")(1), x))
                         .transform(rdd => {
                            rdd.leftOuterJoin(blacksRDD)
                               .filter(x => x._2._2.getOrElse(false) != true)
                               .map(x => x._2._1)
                          })
        
        clicklog.print()
        
        ssc.start()
        ssc.awaitTermination()
      }
    }

## 五、 spark  streaming 整合 spark  sql  实战 ##

**Spark Streaming整合Spark SQL完成词频统计**

object SqlNetworkWordCount {
       def main(args: Array[String]): Unit = {
     
        val sparkConf = new SparkConf().setMaster("local[2]").setAppName("SqlNetworkWordCount")
        val ssc = new StreamingContext(sparkConf, Seconds(5))
        val lines = ssc.socketTextStream("192.168.42.85", 6789)
        val words = lines.flatMap(_.split(" "))
     
        // Convert RDDs of the words DStream to DataFrame and run SQL query
        words.foreachRDD {(rdd: RDD[String], time: Time) =>
          // Get the singleton instance of SparkSession
          val spark = SparkSessionSingleton.getInstance(rdd.sparkContext.getConf)
          import spark.implicits._
          // Convert RDD[String] to RDD[case class] to DataFrame
          val wordsDataFrame = rdd.map(w => Record(w)).toDF()
          // Creates a temporary view using the DataFrame
          wordsDataFrame.createOrReplaceTempView("words")
          // Do word count on table using SQL and print it
          val wordCountsDataFrame =
            spark.sql("select word, count(*) as total from words group by word")
          println(s"========= $time =========")
          wordCountsDataFrame.show()
        }
        ssc.start()
        ssc.awaitTermination()
      }
    }
     
    /** Case class for converting RDD to DataFrame */
    case class Record(word: String)
     
    /** Lazily instantiated singleton instance of SparkSession */
    object SparkSessionSingleton {
      @transient private var instance: SparkSession = _
      def getInstance(sparkConf: SparkConf): SparkSession = {
        if (instance == null) {
          instance = SparkSession
            .builder
            .config(sparkConf)
            .getOrCreate()
        }
        instance
      }
    }

[Spark Streaming]: https://blog.csdn.net/qq_31905135/article/details/85676609
[Spark Streaming 1]: https://blog.csdn.net/qq_45400755/article/details/102736047