spark streaming測(cè)試之三有狀態(tài)的接收數(shù)據(jù)

發(fā)布時(shí)間：2020-06-07 02:00:27 來源：網(wǎng)絡(luò) 閱讀：953 作者：996440550 欄目：大數(shù)據(jù)

測(cè)試思路：

首先，使用上篇文章的程序一發(fā)送網(wǎng)絡(luò)數(shù)據(jù)；

其次，運(yùn)行spark程序，觀察效果。

說明：

1. 這里使用到了更新函數(shù)；

2. 使用檢查點(diǎn)來保證狀態(tài)。

sparkStreaming

import org.apache.log4j.{LoggerLevel}
import org.apache.spark.streaming.{SecondsStreamingContext}
import org.apache.spark.{SparkContextSparkConf}
import org.apache.spark.streaming.StreamingContext._

object StatefulWordCount {
   def main(args:Array[]){

    Logger.().setLevel(Level.)
    Logger.().setLevel(Level.)

    updateFunc = (values: []state:Option[]) => {
      currentCount = values.foldLeft()(_+_)
      previousCount = state.getOrElse()
      (currentCount + previousCount)
    }

    conf = SparkConf().setAppName().setMaster()
    sc = SparkContext(conf)

    ssc = StreamingContext(sc())
    ssc.checkpoint()

    lines = ssc.socketTextStream(args()args().toInt)
    words = lines.flatMap(_.split())
    wordCounts = words.map(x=>(x))

    stateDstream = wordCounts.updateStateByKey[](updateFunc)
    stateDstream.print()
    ssc.start()
    ssc.awaitTermination()
  }
}

向AI問一下細(xì)節(jié)

spark streaming測(cè)試之三有狀態(tài)的接收數(shù)據(jù)

猜你喜歡

最新資訊

相關(guān)推薦

相關(guān)標(biāo)簽