如何利用Spark Streaming實現(xiàn)UV統(tǒng)計

問題描述：首先以一個簡單的示例開始：用Spark Streaming對從TCP連接中接收的文本進(jìn)行單詞計數(shù)。 /** * 功能：用spark streaming實現(xiàn)的針對流式數(shù)據(jù)進(jìn)行單詞計數(shù)的程序。 * 該程序只是對數(shù)據(jù)流中的每一批數(shù)據(jù)進(jìn)行單獨的計數(shù)，而沒有進(jìn)行增量計數(shù)。

回答(1).首先以一個簡單的示例開始：用Spark Streaming對從TCP連接中接收的文本進(jìn)行單詞計數(shù)。 /** * 功能：用spark streaming實現(xiàn)的針對流式數(shù)據(jù)進(jìn)行單詞計數(shù)的程序。 * 該程序只是對數(shù)據(jù)流中的每一批數(shù)據(jù)進(jìn)行單獨的計數(shù)，而沒有進(jìn)行增量計數(shù)。

回答(2).首先以一個簡單的示例開始：用Spark Streaming對從TCP連接中接收的文本進(jìn)行單詞計數(shù)。 /** * 功能：用spark streaming實現(xiàn)的針對流式數(shù)據(jù)進(jìn)行單詞計數(shù)的程序。 * 該程序只是對數(shù)據(jù)流中的每一批數(shù)據(jù)進(jìn)行單獨的計數(shù)，而沒有進(jìn)行增量計數(shù)。 * 環(huán)境：spark 1.6.1, scala 2.10.4 */ // 引入相關(guān)類庫 import org.apache.spark._ import org.apache.spark.streaming._ object NetworkWordCount { def main(args: Array[String]) { // Spark Streaming程序以StreamingContext為起點，其內(nèi)部維持了一個SparkContext的實例。 // 這里我們創(chuàng)建一個帶有兩個本地線程的StreamingContext，并設(shè)置批處理間隔為1秒。 val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount") val ssc = new StreamingContext(conf, Seconds(1)) // 在一個Spark應(yīng)用中默認(rèn)只允許有一個SparkContext，默認(rèn)地spark-shell已經(jīng)為我們創(chuàng)建好了 // SparkContext，名為sc。因此在spark-shell中應(yīng)該以下述方式創(chuàng)建StreamingContext，以 // 避免創(chuàng)建再次創(chuàng)建SparkContext而引起錯誤： // val ssc = new StreamingContext(sc, Seconds(1)) // 創(chuàng)建一個從TCP連接獲取流數(shù)據(jù)的DStream，其每條記錄是一行文本 val lines = ssc.socketTextStream("localhost", 9999) // 對DStream進(jìn)行轉(zhuǎn)換，最終得到計算結(jié)果 val res = lines.flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _) // 打印該DStream中每個RDD中的前十個元素 res.print() // 執(zhí)行完上面代碼，Spark Streaming并沒有真正開始處理數(shù)據(jù)，而只是記錄需在數(shù)據(jù)上執(zhí)行的操作。 // 當(dāng)我們設(shè)置好所有需要在數(shù)據(jù)上執(zhí)行的操作以后，我們就可以開始真正地處理數(shù)據(jù)了。如下： ssc.start() // 開始計算 ssc.awaitTermination() // 等待計算終止 } } 為了測試程序，我們得有TCP數(shù)據(jù)源作為輸入，這可以使用Netcat(一般linux系統(tǒng)中都有，如果是windows系統(tǒng)，則推薦你使用 Ncat ，Ncat是一個改進(jìn)版的Netcat)。如下使......

回答(3).首先，需要將以下代碼編譯成jar包，然后在flume中使用，代碼轉(zhuǎn)自這里 （如果發(fā)現(xiàn)需要依賴的工具類神馬的，請在相同目錄下的scala文件中找一找） package org.apache.spark.streaming.flume.sink import java.net.InetSocketAddress import java....

回答(4).初始化的過程主要可以概括為兩點： 1）調(diào)度器的初始化。 調(diào)度器調(diào)度 Spark Streaming 的運行，用戶可以通過配置相關(guān)參數(shù)進(jìn)行調(diào)優(yōu)。 2）將輸入流的接收器轉(zhuǎn)化為 RDD 在集群進(jìn)行分布式分配，然后啟動接收器集合中的每個接收器。

回答(5).　　初始化的過程主要可以概括為兩點： 　　1）調(diào)度器的初始化。 　　調(diào)度器調(diào)度 Spark Streaming 的運行，用戶可以通過配置相關(guān)參數(shù)進(jìn)行調(diào)優(yōu)。 　　2）將輸入流的接收器轉(zhuǎn)化為 RDD 在集群進(jìn)行分布式分配，然后啟動接收器集合中的每個接收器。 　　針對不同的數(shù)據(jù)源， Spark Streaming 提供了不同的數(shù)據(jù)接收器，分布在各個節(jié)點上的每個接收器可以認(rèn)為是一個特定的進(jìn)程，接收一部分流數(shù)據(jù)作為輸入。

回答(6).參看這個貼子 里面說介紹

回答(7).隨著大數(shù)據(jù)的發(fā)展，人們對大數(shù)據(jù)的處理要求也越來越高，原有的批處理框架MapReduce適合離線計算，卻無法滿足實時性要求較高的業(yè)務(wù)，如實時推薦、用戶行為分析等。 Spark Streaming是建立在Spark上的實時計算框架，通過它提供的豐富的API、基于內(nèi).

本文章由三昆廠家整理原創(chuàng)，轉(zhuǎn)載請注明出處：http://www.eaibonline.cn/UVj/3691.html