Spark之Streaming学习（）

Spark Stream简介

• SparkStreaming是一套框架。
• SparkStreaming是Spark核心API的一个扩展，可以实现高吞吐量的，具备容错机制的实时流数据处理。
• 支持多种数据源获取数据：

•  Spark Streaming接收Kafka、Flume、HDFS等各种来源的实时输入数据，进行处理后，处理结构保存在HDFS、DataBase等各种地方。
•  *使用的最多的是kafka+Spark Streaming

• Spark Streaming和SparkCore的关系：

• Spark处理的是批量的数据（离线数据），Spark Streaming实际上处理并不是像Strom一样来一条处理一条数据，而是对接的外部数据流之后按照时间切分，批处理一个个切分后的文件，和Spark处理逻辑是相同的。

2.TCPStram编程小案例

使用 sparkStream 读取tcp的数据，统计单词数前十的单词
注意：

1）spark是以批处理为主，以微批次处理为辅助解决实时处理问题
flink以stream为主，以stram来解决批处理数据
2）Stream的数据过来是需要存储的，默认存储级别：MEMORY_AND_DISK_SER_2
3）因为tcp需要一个线程去接收数据，故至少两个core，
基本的Stream中，只有FileStream没有Receiver，其它都有，而高级的Stream中Kafka选择的是DirStream，也不使用Receiver
4）Stream 一旦启动都不会主动关闭，但是可以通过WEB-UI进行优雅的关闭
5）一旦start()就不要做多余的操作，一旦stop则程序不能重新start，一个程序中只能有一个StreamContext
6）对DStrem做的某个操作就是对每个RDD的操作
7）receiver是运行在excuetor上的作业，该作业会一直一直的运行者，每隔一定时间接收到数据就通知driver去启动作业

package com.wsk.spark.stream
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.PairDStreamFunctions
import org.apache.spark.streaming.{Seconds, StreamingContext}
object TcpStream {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setMaster("local[2]")
      .setAppName("word count")

    //每隔一秒的数据为一个batch
    val ssc = new StreamingContext(conf,Seconds(5))
    //读取的机器以及端口
    val lines = ssc.socketTextStream("192.168.76.120", 1314)
    //对DStrem做的某个操作就是对每个RDD的每个操作
    val words = lines.flatMap(_.split(" "))
    val pair = words.map(word =>(word,1))

    val wordCounts = pair.reduceByKey((x,y)=>x+y)
    // Print the first ten elements of each RDD generated in this DStream to the console
    wordCounts.print()


    ssc.start()  // Start the computation
    ssc.awaitTermination() // Wait for the computation to terminate
  }
}

3.UpdateStateBykey编程小案例

• 通过UpdateStateBykey这个Transformations方法，实现跨批次的wordcount
• 注意：updateStateByKey Transformations，实现跨批次的处理，但是checkpoint会生成很多小文件，生产上，最合理的方式弃用checkpoint，直接写入DB
• 生产上Spark是尽量尽量不使用Checkpoint的，深坑，生成太多小文件了

package com.wsk.spark.stream

import org.apache.spark.SparkConf
import org.apache.spark.HashPartitioner
import org.apache.spark.streaming._
object UpdateStateBykeyTfTest {
  def main(args: Array[String]) {

    ///函数的返回类型是Some(Int)，因为preValue的类型就是Option
    ///函数的功能是将当前时间间隔内产生的Key的value集合的和，与之前的值相加
    val updateFunc = (values: Seq[Int], preValue: Option[Int]) => {
      val currentCount = values.sum
      val previousCount = preValue.getOrElse(0)
      Some(currentCount + previousCount)
    }

    ///入参是三元组遍历器，三个元组分别表示Key、当前时间间隔内产生的对应于Key的Value集合、上一个时间点的状态
    ///newUpdateFunc的返回值要求是iterator[(String,Int)]类型的
    val newUpdateFunc = (iterator: Iterator[(String, Seq[Int], Option[Int])]) => {
      ///对每个Key调用updateFunc函数(入参是当前时间间隔内产生的对应于Key的Value集合、上一个时间点的状态）得到最新状态
      ///然后将最新状态映射为Key和最新状态
      iterator.flatMap(t => updateFunc(t._2, t._3).map(s => (t._1, s)))
    }

    val sparkConf = new SparkConf()
      .setAppName("StatefulNetworkWordCount")
      .setMaster("local[3]")
    // Create the context with a 5 second batch size
    val ssc = new StreamingContext(sparkConf, Seconds(5))
    ssc.checkpoint(".")

    // Initial RDD input to updateStateByKey
    val initialRDD = ssc.sparkContext.parallelize(List(("hello", 1), ("world", 1)))
    // Create a ReceiverInputDStream on target ip:port and count the
    // words in input stream of 
 delimited test (eg. generated by 'nc')
    val lines = ssc.socketTextStream("192.168.76.120", 1314)
    val words = lines.flatMap(_.split(" "))
    val wordDstream = words.map(x => (x, 1))

    // Update the cumulative count using updateStateByKey
    // This will give a Dstream made of state (which is the cumulative count of the words)
    //注意updateStateByKey的四个参数，第一个参数是状态更新函数
//    val stateDstream = wordDstream.updateStateByKey[Int](newUpdateFunc,
//      new HashPartitioner(ssc.sparkContext.defaultParallelism), true, initialRDD)
    val stateDstream = wordDstream.updateStateByKey(updateFunc)

    stateDstream.print()
    ssc.start()
    ssc.awaitTermination()
  }
}