流式計算應用12

計算應用學習流式01有狀態(tài)轉(zhuǎn)化介紹目錄02UpdateStateByKey有狀態(tài)轉(zhuǎn)化介紹1有狀態(tài)轉(zhuǎn)化介紹

DatehelloDateDateDateDateDatehelloDateDateDateDate3s3s采集器DStreamHello,2UpdateStateByKey2UpdateStateByKey

UpdateStateByKey原語用于記錄歷史記錄，有時，我們需要在DStream中跨批次維護狀態(tài)(例如流計算中累加wordcount)。針對這種情況，updateStateByKey()為我們提供了對一個狀態(tài)變量的訪問，用于鍵值對形式的DStream。給定一個由(鍵，事件)對構(gòu)成的DStream，并傳遞一個指定如何根據(jù)新的事件更新每個鍵對應狀態(tài)的函數(shù)，它可以構(gòu)建出一個新的DStream，其內(nèi)部數(shù)據(jù)為(鍵，狀態(tài))對。updateStateByKey()的結(jié)果會是一個新的DStream，其內(nèi)部的RDD序列是由每個時間區(qū)間對應的(鍵，狀態(tài))對組成的。updateStateByKey操作使得我們可以在用新信息進行更新時保持任意的狀態(tài)。為使用這個功能，需要做下面兩步:1.定義狀態(tài)，狀態(tài)可以是一個任意的數(shù)據(jù)類型。2.定義狀態(tài)更新函數(shù)，用此函數(shù)闡明如何使用之前的狀態(tài)和來自輸入流的新值對狀態(tài)進行更新。使用updateStateByKey需要對檢查點目錄進行配置，會使用檢查點來保存狀態(tài)。

UpdateStateByKey

Wordcount：編寫代碼objectSparkStreaming05_State{

defmain(args:Array[String]):Unit={

valsparkConf=newSparkConf().setMaster("local[*]").setAppName("SparkStreaming")valssc=newStreamingContext(sparkConf,Seconds(3))ssc.checkpoint("cp")

//無狀態(tài)數(shù)據(jù)操作，只對當前的采集周期內(nèi)的數(shù)據(jù)進行處理//在某些場合下，需要保留數(shù)據(jù)統(tǒng)計結(jié)果（狀態(tài)），實現(xiàn)數(shù)據(jù)的匯總//使用有狀態(tài)操作時，需要設(shè)定檢查點路徑valdatas=ssc.socketTextStream("localhost",9999)

valwordToOne=datas.map((_,1))

//valwordToCount=wordToOne.reduceByKey(_+_)

//updateStateByKey：根據(jù)key對數(shù)據(jù)的狀態(tài)進行更新//傳遞的參數(shù)中含有兩個值//第一個值表示相同的key的value數(shù)據(jù)//第二個值表示緩存區(qū)相同key的value數(shù)據(jù)valstate=wordToOne.updateStateByKey((seq:Seq[Int],buff:Option[Int])=>{valnewCount=buff.getOrElse(0)+seq.sumOption(newCount)})state.print()

ssc.start()ssc.awaitTermination()UpdateStateByKey

Wordcount：結(jié)果展示-------------------------------------------Time:1612426170000ms-------------------------------------------(hello,1)

-------------------------------------------Time:1612426176000ms-------------------------------------------(hello,1)(spark,1)

-------------------------------------------Time:1612426179000