目錄

SparkStreaming實戰(zhàn)案例一 WordCount

需求

準備工作

代碼實現(xiàn)

?第一種方式：構(gòu)建SparkConf對象

?第二種方式：構(gòu)建SparkContext對象

完整代碼如下所示：

應用監(jiān)控

?其一、Streaming流式應用概要信息

?其二、性能衡量標準

---

SparkStreaming實戰(zhàn)案例一 WordCount

需求

從TCP Socket數(shù)據(jù)源實時消費數(shù)據(jù)，對每批次Batch數(shù)據(jù)進行詞頻統(tǒng)計WordCount，流程圖如下：

?

準備工作

1.在node01上安裝nc命令

nc是netcat的簡稱，原本是用來設(shè)置路由器,我們可以利用它向某個端口發(fā)送數(shù)據(jù)

yum install -y nc

2.在node01啟動客戶端工具發(fā)送消息

?nc -lk 9999

?

代碼實現(xiàn)

http://spark.apache.org/docs/latest/streaming-programming-guide.html

從官方文檔可知，提供兩種方式構(gòu)建StreamingContext實例對象，如下：

?第一種方式：構(gòu)建SparkConf對象

?

?

?第二種方式：構(gòu)建SparkContext對象

?

?

完整代碼如下所示：

package cn.itcast.streamingimport org.apache.commons.lang3.StringUtils
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}/*** 基于IDEA集成開發(fā)環(huán)境，編程實現(xiàn)從TCP Socket實時讀取流式數(shù)據(jù)，對每批次中數(shù)據(jù)進行詞頻統(tǒng)計。*/
object SparkStreamingDemo01_WordCount {def main(args: Array[String]): Unit = {val conf: SparkConf = new SparkConf().setAppName(this.getClass.getSimpleName.stripSuffix("$")).setMaster("local[*]")val sc: SparkContext = new SparkContext(conf)sc.setLogLevel("WARN")//batchDuration the time interval at which streaming data will be divided into batches//流數(shù)據(jù)將被劃分為批的時間間隔,就是每隔多久對流數(shù)據(jù)進行一次微批劃分!val ssc: StreamingContext = new StreamingContext(sc, Seconds(5))val inputDStream: ReceiverInputDStream[String] = ssc.socketTextStream("node1", 9999)val resultDStream: DStream[(String, Int)] = inputDStream.filter(StringUtils.isNotBlank(_)).flatMap(_.trim.split("\\s+")).map((_, 1)).reduceByKey(_ + _)resultDStream.print(10)// 啟動并等待程序停止// 對于流式應用來說，需要啟動應用ssc.start()// 流式應用啟動以后，正常情況一直運行（接收數(shù)據(jù)、處理數(shù)據(jù)和輸出數(shù)據(jù)），除非人為終止程序或者程序異常停止ssc.awaitTermination()// 關(guān)閉流式應用(參數(shù)一：是否關(guān)閉SparkContext，參數(shù)二：是否優(yōu)雅的關(guān)閉）ssc.stop(stopSparkContext = true, stopGracefully = true)//注意://上面的代碼可以做WordCount,但是只能對當前批次的數(shù)據(jù)進行累加!}
}

?

?

應用監(jiān)控

運行上述詞頻統(tǒng)計案例，登錄到WEB UI監(jiān)控頁面：http://localhost:4040/

查看相關(guān)監(jiān)控信息。

?其一、Streaming流式應用概要信息

運行結(jié)果監(jiān)控截圖：

?

?

?

每批次Batch數(shù)據(jù)處理總時間TD = 批次調(diào)度延遲時間SD + 批次數(shù)據(jù)處理時間PT

?

?其二、性能衡量標準

SparkStreaming實時處理數(shù)據(jù)性能如何（是否可以實時處理數(shù)據(jù)）？？如何衡量的呢？？

需要滿足：

每批次數(shù)據(jù)處理時間TD ?<= ?BatchInterval每批次時間間隔

?

總結(jié)

以上是生活随笔為你收集整理的2021年大数据Spark（三十六）：SparkStreaming实战案例一 WordCount的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

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