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 \n 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() } }