場景描述：這是一個Spark的實戰題目，也是在面試中經常出現的一類題目。

問題描述

對于一個大型網站，用戶訪問量嘗嘗高達數十億。對于數十億是一個什么樣的概念，我們這里可以簡單的計算一下。對于一個用戶，單次訪問，我們通常會記錄下哪些數據呢？

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

我們單單從用戶id來說，比如10011802330414，這個ID，那么我們一個id差不多就是一個long類型，因為在大量數據存儲的時候，我們都是采用文本存儲。因此對于5億個用戶ID，完全存儲在磁盤當中，大概是5G的大小，對于這個大小，并不能算是大數據。但是對于一個案例來說，已經非常足夠了。

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

當然要解決這個問題，你可以依然在local模式下運行項目，但是你得有足夠的磁盤空間和內存空間，大概8G磁盤空間（因為除了數據本身，spark運行過程還要產生一些臨時數據），5G內存（要進行reduceByKey）。為了真正展示spark的特性，我們這個案例，將會運行在spark集群上。

關于如何搭建集群，我準備在后續的章節補上。但是在網上有大量的集群搭建教程，其中不乏一些詳細優秀的教程。當然，這節我們不講如何搭建集群，但是我們仍然可以開始我們的案例。

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

5億條ID數據，首先可以用map將其緩存到RDD中，然后對RDD進行reduceByKey，最后找出出現最多的ID。思路很簡單，因此代碼量也不會很多。

實現

scala實現
首先是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)//生成一個0-9999的列表val list = 1 until 10000val id =new RandomId()//這里記錄最大的次數var max = 0//這里記錄最大次數的IDvar maxId = 0Lval lastNum = sc.parallelize(list)//第一步生成5億條數據.flatMap(num => {//遍歷list列表//總共遍歷1萬次每次生成5萬個IDvar list2 = List(id.next())for (i <- 1 to 50000){list2 = id.next() :: list2}//這里記錄當前生成ID的百分比println(num/1000.0 +"%")//返回生成完成后的list//每次循環里面都包含5萬個IDlist2})//遍歷5億條數據//為每條數據出現標記1.map((_,1))//對標記后的數據進行處理//得到每個ID出現的次數，即（ID，Count）.reduceByKey(_+_)//遍歷處理后的數據.foreach(x => {//將最大值存儲在max中if (x._2 > max){max = x._2maxId = x._1//若X比之前記錄的值大，則輸出該id和次數//最后一次輸出結果，則是出現次數最多的的ID和以及其出現的次數//當然出現次數最多的可能有多個ID//這里只輸出一個println(x)}})}}

然后是用它生成5億條數據

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億條數據

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)//生成一個0-9999的列表val list = 1 until 10000val id =new RandomId()//這里記錄最大的次數var max = 0//這里記錄最大次數的IDvar maxId = 0Lval lastNum = sc.parallelize(list)//第一步生成5億條數據.flatMap(num => {//遍歷list列表//總共遍歷1萬次每次生成5萬個IDvar list2 = List(id.next())for (i <- 1 to 50000){list2 = id.next() :: list2}//這里記錄當前生成ID的百分比println(num/1000.0 +"%")//返回生成完成后的list//每次循環里面都包含5萬個IDlist2})//遍歷5億條數據//為每條數據出現標記1.map((_,1))//對標記后的數據進行處理//得到每個ID出現的次數，即（ID，Count）.reduceByKey(_+_)//遍歷處理后的數據.foreach(x => {//將最大值存儲在max中if (x._2 > max){max = x._2maxId = x._1//若X比之前記錄的值大，則輸出該id和次數//最后一次輸出結果，則是出現次數最多的的ID和以及其出現的次數//當然出現次數最多的可能有多個ID//這里只輸出一個println(x)}})} }

運行得到結果
將其提交到spark上運行，觀察日志

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%... ... ...

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

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的數據溢寫到磁盤中。這是由于spark在處理的過程中，由于數據量過于龐大，因此將多的數據溢寫到磁盤，當再次用到時，會從磁盤讀取。對于實時性操作的程序來說，多次、大量讀寫磁盤是絕對不被允許的。但是在處理大數據中，溢寫到磁盤是非常常見的操作。

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

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

結果

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

整個過程持續了將近47min，當然在龐大的集群中，時間能夠大大縮短，要知道，我們現在只采用了4個節點。

我們看到了次數2、4、6、8居然分別出現了兩次，這并不奇怪，因為集群并行運行，異步操作，出現重復結果十分正常，當然我們也可以用并發機制，去處理這個現象。這個在后續的案例中，我們會繼續優化結果。

從結果上看，我們發現5億條數據中，出現最多的ID也僅僅出現了8次，這說明了在大量數據中，很多ID可能只出現了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)//生成一個0-9999的列表val list = 1 until 10000val id =new RandomId()//這里記錄最大的次數var max = 0//這里記錄最大次數的IDvar maxId = 0Lval lastNum = sc.parallelize(list)//第一步生成5億條數據.flatMap(num => {//遍歷list列表//總共遍歷1萬次每次生成5萬個IDvar list2 = List(id.next())for (i <- 1 to 50000){list2 = id.next() :: list2}//這里記錄當前生成ID的百分比println(num/1000.0 +"%")//返回生成完成后的list//每次循環里面都包含5萬個IDlist2})//遍歷5億條數據//為每條數據出現標記1.map((_,1))//對標記后的數據進行處理//得到每個ID出現的次數，即（ID，Count）.reduceByKey(_+_)//為數據進行排序//倒序.sortByKey(false)//次數最多的，在第一個，將其輸出println(lastNum.first())} }

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

根據我們對排序算法的了解，這樣一個龐大數據集進行排序，勢必要耗費大量資源。因此，我們能夠容忍輸出一些冗余信息，但不影響我們的得到正確結果。

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

總結

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

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