場(chǎng)景描述：這是一個(gè)Spark的實(shí)戰(zhàn)題目，也是在面試中經(jīng)常出現(xiàn)的一類題目。

問題描述

對(duì)于一個(gè)大型網(wǎng)站，用戶訪問量嘗嘗高達(dá)數(shù)十億。對(duì)于數(shù)十億是一個(gè)什么樣的概念，我們這里可以簡(jiǎn)單的計(jì)算一下。對(duì)于一個(gè)用戶，單次訪問，我們通常會(huì)記錄下哪些數(shù)據(jù)呢？

1、用戶的id
2、用戶訪問的時(shí)間
3、用戶逗留的時(shí)間
4、用戶執(zhí)行的操作
5、用戶的其余數(shù)據(jù)（比如IP等等）

我們單單從用戶id來說，比如10011802330414，這個(gè)ID，那么我們一個(gè)id差不多就是一個(gè)long類型，因?yàn)樵诖罅繑?shù)據(jù)存儲(chǔ)的時(shí)候，我們都是采用文本存儲(chǔ)。因此對(duì)于5億個(gè)用戶ID，完全存儲(chǔ)在磁盤當(dāng)中，大概是5G的大小，對(duì)于這個(gè)大小，并不能算是大數(shù)據(jù)。但是對(duì)于一個(gè)案例來說，已經(jīng)非常足夠了。

我們會(huì)產(chǎn)生一個(gè)5億條ID的數(shù)據(jù)集，我們上面說到，這個(gè)數(shù)據(jù)集大小為5G（不壓縮的情況下），因此我不會(huì)在GitHub上上傳這樣一個(gè)數(shù)據(jù)集，但是我們提供一個(gè)方法，來生成一個(gè)5億條數(shù)據(jù)。

當(dāng)然要解決這個(gè)問題，你可以依然在local模式下運(yùn)行項(xiàng)目，但是你得有足夠的磁盤空間和內(nèi)存空間，大概8G磁盤空間（因?yàn)槌藬?shù)據(jù)本身，spark運(yùn)行過程還要產(chǎn)生一些臨時(shí)數(shù)據(jù)），5G內(nèi)存（要進(jìn)行reduceByKey）。為了真正展示spark的特性，我們這個(gè)案例，將會(huì)運(yùn)行在spark集群上。

關(guān)于如何搭建集群，我準(zhǔn)備在后續(xù)的章節(jié)補(bǔ)上。但是在網(wǎng)上有大量的集群搭建教程，其中不乏一些詳細(xì)優(yōu)秀的教程。當(dāng)然，這節(jié)我們不講如何搭建集群，但是我們?nèi)匀豢梢蚤_始我們的案例。

問題分析
那么現(xiàn)在我們擁有了一個(gè)5億條數(shù)據(jù)（實(shí)際上這個(gè)數(shù)據(jù)并不以文本存儲(chǔ)，而是在運(yùn)行的時(shí)候生成），從五億條數(shù)據(jù)中，找出訪問次數(shù)最多的人，這看起來并不難。但實(shí)際上我們想要通過這個(gè)案例了解spark的真正優(yōu)勢(shì)。

5億條ID數(shù)據(jù)，首先可以用map將其緩存到RDD中，然后對(duì)RDD進(jìn)行reduceByKey，最后找出出現(xiàn)最多的ID。思路很簡(jiǎn)單，因此代碼量也不會(huì)很多。

實(shí)現(xiàn)

scala實(shí)現(xiàn)
首先是ID生成方法：
RandomId.class

import org.apache.spark.{SparkConf, SparkContext}object ActiveVisitor {def main(args: Array[String]): Unit = {val conf = new SparkConf().setMaster("spark://master:7077").setAppName("ActiveVisitor")val sc = new SparkContext(conf)//生成一個(gè)0-9999的列表val list = 1 until 10000val id =new RandomId()//這里記錄最大的次數(shù)var max = 0//這里記錄最大次數(shù)的IDvar maxId = 0Lval lastNum = sc.parallelize(list)//第一步生成5億條數(shù)據(jù).flatMap(num => {//遍歷list列表//總共遍歷1萬次每次生成5萬個(gè)IDvar list2 = List(id.next())for (i <- 1 to 50000){list2 = id.next() :: list2}//這里記錄當(dāng)前生成ID的百分比println(num/1000.0 +"%")//返回生成完成后的list//每次循環(huán)里面都包含5萬個(gè)IDlist2})//遍歷5億條數(shù)據(jù)//為每條數(shù)據(jù)出現(xiàn)標(biāo)記1.map((_,1))//對(duì)標(biāo)記后的數(shù)據(jù)進(jìn)行處理//得到每個(gè)ID出現(xiàn)的次數(shù)，即（ID，Count）.reduceByKey(_+_)//遍歷處理后的數(shù)據(jù).foreach(x => {//將最大值存儲(chǔ)在max中if (x._2 > max){max = x._2maxId = x._1//若X比之前記錄的值大，則輸出該id和次數(shù)//最后一次輸出結(jié)果，則是出現(xiàn)次數(shù)最多的的ID和以及其出現(xiàn)的次數(shù)//當(dāng)然出現(xiàn)次數(shù)最多的可能有多個(gè)ID//這里只輸出一個(gè)println(x)}})}}

然后是用它生成5億條數(shù)據(jù)

import org.apache.spark.{SparkConf, SparkContext}object ActiveVisitor {def main(args: Array[String]): Unit = {val conf = new SparkConf().setMaster("spark://master:7077").setAppName("ActiveVisitor")val sc = new SparkContext(conf)val list = 1 until 100000val id =new RandomId()var max = 0var maxId = 0Lval lastNum = sc.parallelize(list).flatMap(num => {var list2 = List(id.next())for (i <- 1 to 50000){list2 = id.next() :: list2}println(num +"%")list2}).map((_,1)).reduceByKey(_+_).foreach(x => {if (x._2 > max){max = x._2maxId = x._1println(x)}})} }

處理5億條數(shù)據(jù)

import org.apache.spark.{SparkConf, SparkContext}object ActiveVisitor {def main(args: Array[String]): Unit = {val conf = new SparkConf().setMaster("spark://master:7077").setAppName("ActiveVisitor")val sc = new SparkContext(conf)//生成一個(gè)0-9999的列表val list = 1 until 10000val id =new RandomId()//這里記錄最大的次數(shù)var max = 0//這里記錄最大次數(shù)的IDvar maxId = 0Lval lastNum = sc.parallelize(list)//第一步生成5億條數(shù)據(jù).flatMap(num => {//遍歷list列表//總共遍歷1萬次每次生成5萬個(gè)IDvar list2 = List(id.next())for (i <- 1 to 50000){list2 = id.next() :: list2}//這里記錄當(dāng)前生成ID的百分比println(num/1000.0 +"%")//返回生成完成后的list//每次循環(huán)里面都包含5萬個(gè)IDlist2})//遍歷5億條數(shù)據(jù)//為每條數(shù)據(jù)出現(xiàn)標(biāo)記1.map((_,1))//對(duì)標(biāo)記后的數(shù)據(jù)進(jìn)行處理//得到每個(gè)ID出現(xiàn)的次數(shù)，即（ID，Count）.reduceByKey(_+_)//遍歷處理后的數(shù)據(jù).foreach(x => {//將最大值存儲(chǔ)在max中if (x._2 > max){max = x._2maxId = x._1//若X比之前記錄的值大，則輸出該id和次數(shù)//最后一次輸出結(jié)果，則是出現(xiàn)次數(shù)最多的的ID和以及其出現(xiàn)的次數(shù)//當(dāng)然出現(xiàn)次數(shù)最多的可能有多個(gè)ID//這里只輸出一個(gè)println(x)}})} }

運(yùn)行得到結(jié)果
將其提交到spark上運(yùn)行，觀察日志

1% 5000% 2% 5001% 3% 5002% 4% 5003% 5% 5004% 6% 5005% 7% 5006% 8% 5007% 9% 5008% 10% 5009% 11% 5010% 12% 5011% 5012% 13% 5013% 14% 15% 5014%... ... ...

這里是輸出的部分日志，從日志中，我們顯然發(fā)現(xiàn)，程序是并行的。我采用的集群由四個(gè)節(jié)點(diǎn)組成，每個(gè)節(jié)點(diǎn)提供5G的內(nèi)存空間，集群在不同節(jié)點(diǎn)中運(yùn)行，有節(jié)點(diǎn)分配到的分區(qū)是從1開始，而有節(jié)點(diǎn)則是從5000開始，因此程序并沒有按照我們所想的從1%-9999%。好在未按照順序執(zhí)行，也并不影響最終結(jié)果，畢竟最終要進(jìn)行一個(gè)reduceByKey，才是我們真正需要得到結(jié)果的地方。
再看日志另一部分：

5634% 5635% 5636% 5637% 5638% 5639% 5640% 5641% 5642% 5643% 5644% 5645% 2019-03-05 11:52:14 INFO ExternalSorter:54 - Thread 63 spilling in-memory map of 1007.3 MB to disk (2 times so far) 647% 648% 649% 650% 651% 652% 653% 654% 655% 656%

注意到這里，spilling in-memory map of 1007.3 MB to disk，spilling操作將map中的 1007.3 MB的數(shù)據(jù)溢寫到磁盤中。這是由于spark在處理的過程中，由于數(shù)據(jù)量過于龐大，因此將多的數(shù)據(jù)溢寫到磁盤，當(dāng)再次用到時(shí)，會(huì)從磁盤讀取。對(duì)于實(shí)時(shí)性操作的程序來說，多次、大量讀寫磁盤是絕對(duì)不被允許的。但是在處理大數(shù)據(jù)中，溢寫到磁盤是非常常見的操作。

事實(shí)上，在完整的日志中，我們可以看到有相當(dāng)一部分日志是在溢寫磁盤的時(shí)候生成的，大概49次（這是我操作過程中的總數(shù)）
如圖：

總共出現(xiàn)49條溢寫操作的日志，每次大概是1G，這也印證了我們5億條數(shù)據(jù)，占據(jù)空間5G的一個(gè)說法。事實(shí)上，我曾將這5億條數(shù)據(jù)存儲(chǔ)在磁盤中，的確其占據(jù)的空間是5G左右。

結(jié)果

最終，我們可以在日志中看到結(jié)果。

整個(gè)過程持續(xù)了將近47min，當(dāng)然在龐大的集群中，時(shí)間能夠大大縮短，要知道，我們現(xiàn)在只采用了4個(gè)節(jié)點(diǎn)。

我們看到了次數(shù)2、4、6、8居然分別出現(xiàn)了兩次，這并不奇怪，因?yàn)榧翰⑿羞\(yùn)行，異步操作，出現(xiàn)重復(fù)結(jié)果十分正常，當(dāng)然我們也可以用并發(fā)機(jī)制，去處理這個(gè)現(xiàn)象。這個(gè)在后續(xù)的案例中，我們會(huì)繼續(xù)優(yōu)化結(jié)果。

從結(jié)果上看，我們發(fā)現(xiàn)5億條數(shù)據(jù)中，出現(xiàn)最多的ID也僅僅出現(xiàn)了8次，這說明了在大量數(shù)據(jù)中，很多ID可能只出現(xiàn)了1次、2次。這也就是為什么最后我采用的是foreach方法去尋找最大值，而不采用如下的方法

import org.apache.spark.{SparkConf, SparkContext}object ActiveVisitor {def main(args: Array[String]): Unit = {val conf = new SparkConf().setMaster("spark://master:7077").setAppName("ActiveVisitor")val sc = new SparkContext(conf)//生成一個(gè)0-9999的列表val list = 1 until 10000val id =new RandomId()//這里記錄最大的次數(shù)var max = 0//這里記錄最大次數(shù)的IDvar maxId = 0Lval lastNum = sc.parallelize(list)//第一步生成5億條數(shù)據(jù).flatMap(num => {//遍歷list列表//總共遍歷1萬次每次生成5萬個(gè)IDvar list2 = List(id.next())for (i <- 1 to 50000){list2 = id.next() :: list2}//這里記錄當(dāng)前生成ID的百分比println(num/1000.0 +"%")//返回生成完成后的list//每次循環(huán)里面都包含5萬個(gè)IDlist2})//遍歷5億條數(shù)據(jù)//為每條數(shù)據(jù)出現(xiàn)標(biāo)記1.map((_,1))//對(duì)標(biāo)記后的數(shù)據(jù)進(jìn)行處理//得到每個(gè)ID出現(xiàn)的次數(shù)，即（ID，Count）.reduceByKey(_+_)//為數(shù)據(jù)進(jìn)行排序//倒序.sortByKey(false)//次數(shù)最多的，在第一個(gè)，將其輸出println(lastNum.first())} }

這個(gè)方法中，我們對(duì)reduceByKey結(jié)果進(jìn)行排序，輸出排序結(jié)果的第一個(gè)，即次數(shù)最大的ID。這樣做似乎更符合我們的要求。但是實(shí)際上，為了得到同樣的結(jié)果，這樣做，會(huì)消耗更多的資源。如我們所說，很多ID啟其實(shí)只出現(xiàn)了一次，兩次，排序的過程中，仍然要對(duì)其進(jìn)行排序。要知道，由于很多ID只出現(xiàn)一次，排序的數(shù)據(jù)集大小很有可能是數(shù)億的條目。

根據(jù)我們對(duì)排序算法的了解，這樣一個(gè)龐大數(shù)據(jù)集進(jìn)行排序，勢(shì)必要耗費(fèi)大量資源。因此，我們能夠容忍輸出一些冗余信息，但不影響我們的得到正確結(jié)果。

至此，我們完成了5億數(shù)據(jù)中，找出最多出現(xiàn)次數(shù)的數(shù)據(jù)。如果感興趣，可以嘗試用這個(gè)方法解決50億條數(shù)據(jù)，出現(xiàn)最多的數(shù)據(jù)條目。但是這樣做的話，你得準(zhǔn)備好50G的空間。盡管用上述的程序，屬于閱后即焚，但是50億數(shù)據(jù)仍然會(huì)耗費(fèi)大量的時(shí)間。

總結(jié)

以上是生活随笔為你收集整理的Spark的实战题目——寻找5亿次访问中，访问次数最多的人的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。

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