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