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

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

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

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

回答(3).首先，需要將以下代碼編譯成jar包，然后在flume中使用，代碼轉(zhuǎn)自這里 （如果發(fā)現(xiàn)需要依賴的工具類神馬的，請(qǐng)?jiān)谙嗤夸浵碌膕cala文件中找一找） package org.apache.spark.streaming.flume.sink import java.net.InetSocketAddress import java....

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

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

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

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

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