目錄

SparkStreaming實戰案例一 WordCount

需求

準備工作

代碼實現

?第一種方式：構建SparkConf對象

?第二種方式：構建SparkContext對象

完整代碼如下所示：

應用監控

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

?其二、性能衡量標準

---

SparkStreaming實戰案例一 WordCount

需求

從TCP Socket數據源實時消費數據，對每批次Batch數據進行詞頻統計WordCount，流程圖如下：

?

準備工作

1.在node01上安裝nc命令

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

yum install -y nc

2.在node01啟動客戶端工具發送消息

?nc -lk 9999

?

代碼實現

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

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

?第一種方式：構建SparkConf對象

?

?

?第二種方式：構建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集成開發環境，編程實現從TCP Socket實時讀取流式數據，對每批次中數據進行詞頻統計。*/
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//流數據將被劃分為批的時間間隔,就是每隔多久對流數據進行一次微批劃分!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()// 流式應用啟動以后，正常情況一直運行（接收數據、處理數據和輸出數據），除非人為終止程序或者程序異常停止ssc.awaitTermination()// 關閉流式應用(參數一：是否關閉SparkContext，參數二：是否優雅的關閉）ssc.stop(stopSparkContext = true, stopGracefully = true)//注意://上面的代碼可以做WordCount,但是只能對當前批次的數據進行累加!}
}

?

?

應用監控

運行上述詞頻統計案例，登錄到WEB UI監控頁面：http://localhost:4040/

查看相關監控信息。

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

運行結果監控截圖：

?

?

?

每批次Batch數據處理總時間TD = 批次調度延遲時間SD + 批次數據處理時間PT

?

?其二、性能衡量標準

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

需要滿足：

每批次數據處理時間TD ?<= ?BatchInterval每批次時間間隔

?

總結

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

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