/** @Title: SparkStreamingStatefulDemo* @ProjectName spark-scala* @Description: TODO* @author Mr.lu* @date 2018/11/19:10:25*/ /*** 實時處理數(shù)據(jù) 有狀態(tài)計算 需要還原點*/import org.apache.spark._ import org.apache.spark.streaming._ // not necessary since Spark 1.3 object SparkStreamingStatefulDemo {def functionToCreateContext(): StreamingContext = {//streaming至少兩個線程val conf = new SparkConf().setMaster("local[*]").setAppName("streaming")//時間用于分割數(shù)據(jù):Seconds(5)---5秒val ssc = new StreamingContext(conf, Seconds(5))//模擬從tcp端口讀取數(shù)據(jù)//設置還原點目錄ssc.checkpoint("D:\\test\\checkpoint")val ds = ssc.socketTextStream("localhost", 999)//current:rdd的1放在里面 Option:可以沒數(shù)據(jù)（第一次的時候沒有數(shù)據(jù)）def update(current: Seq[Int], old: Option[Int]) = {val newValue = current.sum//第一次key出現(xiàn)時，沒有狀態(tài)，需要初始化val oldValue = old.getOrElse(0)//Some：Option下面的子類---hash--->mapSome(newValue + oldValue)}ds.map(word => (word, 1))//update _:將方法轉成函數(shù) 方法名 _.updateStateByKey(update _).print()ssc}def main(args: Array[String]): Unit = {// //streaming至少兩個線程// val conf=new SparkConf().setMaster("local[*]").setAppName("streaming")// //時間用于分割數(shù)據(jù):Seconds(5)---5秒// val ssc=new StreamingContext(conf,Seconds(5))val ssc = StreamingContext.getOrCreate("D:\\test\\checkpoint", functionToCreateContext _)//啟動streaming context，防止沒有數(shù)據(jù)關閉//如果沒有接受導數(shù)據(jù)，也不會立刻關閉，會嘗試一段時間強制關閉ssc.start()ssc.awaitTermination()} }

轉載于:https://www.cnblogs.com/pigdata/p/10305569.html

總結

以上是生活随笔為你收集整理的SparkStreamingStateful的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內容還不錯，歡迎將生活随笔推薦給好友。