[[243244]]

1、kafka的message包括哪些信息

一個Kafka的Message由一個固定長度的header和一個變長的消息體body組成

header部分由一個字節的magic(文件格式)和四個字節的CRC32(用于判斷body消息體是否正常)構成。

當magic的值為1的時候，會在magic和crc32之間多一個字節的數據：attributes(保存一些相關屬性，

比如是否壓縮、壓縮格式等等);如果magic的值為0，那么不存在attributes屬性

body是由N個字節構成的一個消息體，包含了具體的key/value消息

2、怎么查看kafka的offset

0.9版本以上，可以用最新的Consumer client 客戶端，有consumer.seekToEnd() / consumer.position() 可以用于得到當前最新的offset：

3、hadoop的shuffle過程

一、Map端的shuffle

Map端會處理輸入數據并產生中間結果，這個中間結果會寫到本地磁盤，而不是HDFS。每個Map的輸出會先寫到內存緩沖區中，當寫入的數據達到設定的閾值時，系統將會啟動一個線程將緩沖區的數據寫到磁盤，這個過程叫做spill。

在spill寫入之前，會先進行二次排序，首先根據數據所屬的partition進行排序，然后每個partition中的數據再按key來排序。partition的目是將記錄劃分到不同的Reducer上去，以期望能夠達到負載均衡，以后的Reducer就會根據partition來讀取自己對應的數據。接著運行combiner(如果設置了的話)，combiner的本質也是一個Reducer，其目的是對將要寫入到磁盤上的文件先進行一次處理，這樣，寫入到磁盤的數據量就會減少。最后將數據寫到本地磁盤產生spill文件(spill文件保存在{mapred.local.dir}指定的目錄中，Map任務結束后就會被刪除)。

最后，每個Map任務可能產生多個spill文件，在每個Map任務完成前，會通過多路歸并算法將這些spill文件歸并成一個文件。至此，Map的shuffle過程就結束了。

二、Reduce端的shuffle

Reduce端的shuffle主要包括三個階段，copy、sort(merge)和reduce。

首先要將Map端產生的輸出文件拷貝到Reduce端，但每個Reducer如何知道自己應該處理哪些數據呢?因為Map端進行partition的時候，實際上就相當于指定了每個Reducer要處理的數據(partition就對應了Reducer)，所以Reducer在拷貝數據的時候只需拷貝與自己對應的partition中的數據即可。每個Reducer會處理一個或者多個partition，但需要先將自己對應的partition中的數據從每個Map的輸出結果中拷貝過來。

接下來就是sort階段，也成為merge階段，因為這個階段的主要工作是執行了歸并排序。從Map端拷貝到Reduce端的數據都是有序的，所以很適合歸并排序。最終在Reduce端生成一個較大的文件作為Reduce的輸入。

最后就是Reduce過程了，在這個過程中產生了最終的輸出結果，并將其寫到HDFS上。

4、spark集群運算的模式

Spark 有很多種模式，最簡單就是單機本地模式，還有單機偽分布式模式，復雜的則運行在集群中，目前能很好的運行在 Yarn和 Mesos 中，當然 Spark 還有自帶的 Standalone 模式，對于大多數情況 Standalone 模式就足夠了，如果企業已經有 Yarn 或者 Mesos 環境，也是很方便部署的。

• standalone(集群模式)：典型的Mater/slave模式，不過也能看出Master是有單點故障的;Spark支持ZooKeeper來實現 HA
• on yarn(集群模式)： 運行在 yarn 資源管理器框架之上，由 yarn 負責資源管理，Spark 負責任務調度和計算
• on mesos(集群模式)： 運行在 mesos 資源管理器框架之上，由 mesos 負責資源管理，Spark 負責任務調度和計算
• on cloud(集群模式)：比如 AWS 的 EC2，使用這個模式能很方便的訪問 Amazon的 S3;Spark 支持多種分布式存儲系統：HDFS 和 S3

5、HDFS讀寫數據的過程

讀：

1. 跟namenode通信查詢元數據，找到文件塊所在的datanode服務器
2. 挑選一臺datanode(就近原則，然后隨機)服務器，請求建立socket流
3. datanode開始發送數據(從磁盤里面讀取數據放入流，以packet為單位來做校驗)
4. 客戶端以packet為單位接收，現在本地緩存，然后寫入目標文件

寫：

1. 根namenode通信請求上傳文件，namenode檢查目標文件是否已存在，父目錄是否存在
2. namenode返回是否可以上傳
3. client請求第一個 block該傳輸到哪些datanode服務器上
4. namenode返回3個datanode服務器ABC
5. client請求3臺dn中的一臺A上傳數據(本質上是一個RPC調用，建立pipeline)，A收到請求會繼續調用B，然后B調用C，將真個pipeline建立完成，逐級返回客戶端
6. client開始往A上傳第一個block(先從磁盤讀取數據放到一個本地內存緩存)，以packet為單位，A收到一個packet就會傳給B，B傳給C;A每傳一個packet會放入一個應答隊列等待應答
7. 當一個block傳輸完成之后，client再次請求namenode上傳第二個block的服務器。

6、RDD中reduceBykey與groupByKey哪個性能好，為什么

• reduceByKey：reduceByKey會在結果發送至reducer之前會對每個mapper在本地進行merge，有點類似于在MapReduce中的combiner。這樣做的好處在于，在map端進行一次reduce之后，數據量會大幅度減小，從而減小傳輸，保證reduce端能夠更快的進行結果計算。
• groupByKey：groupByKey會對每一個RDD中的value值進行聚合形成一個序列(Iterator)，此操作發生在reduce端，所以勢必會將所有的數據通過網絡進行傳輸，造成不必要的浪費。同時如果數據量十分大，可能還會造成OutOfMemoryError。

通過以上對比可以發現在進行大量數據的reduce操作時候建議使用reduceByKey。不僅可以提高速度，還是可以防止使用groupByKey造成的內存溢出問題。

7、spark2.0的了解

• 更簡單：ANSI SQL與更合理的API
• 速度更快：用Spark作為編譯器
• 更智能：Structured Streaming

8、rdd 怎么分區寬依賴和窄依賴

• 寬依賴：父RDD的分區被子RDD的多個分區使用 例如 groupByKey、reduceByKey、sortByKey等操作會產生寬依賴，會產生shuffle
• 窄依賴：父RDD的每個分區都只被子RDD的一個分區使用 例如map、filter、union等操作會產生窄依賴

9、spark streaming 讀取kafka數據的兩種方式

這兩種方式分別是：

Receiver-base

使用Kafka的高層次Consumer API來實現。receiver從Kafka中獲取的數據都存儲在Spark Executor的內存中，然后Spark Streaming啟動的job會去處理那些數據。然而，在默認的配置下，這種方式可能會因為底層的失敗而丟失數據。如果要啟用高可靠機制，讓數據零丟失，就必須啟用Spark Streaming的預寫日志機制(Write Ahead Log，WAL)。該機制會同步地將接收到的Kafka數據寫入分布式文件系統(比如HDFS)上的預寫日志中。所以，即使底層節點出現了失敗，也可以使用預寫日志中的數據進行恢復。

Direct

Spark1.3中引入Direct方式，用來替代掉使用Receiver接收數據，這種方式會周期性地查詢Kafka，獲得每個topic+partition的最新的offset，從而定義每個batch的offset的范圍。當處理數據的job啟動時，就會使用Kafka的簡單consumer api來獲取Kafka指定offset范圍的數據。

10、kafka的數據存在內存還是磁盤

Kafka最核心的思想是使用磁盤，而不是使用內存，可能所有人都會認為，內存的速度一定比磁盤快，我也不例外。在看了Kafka的設計思想，查閱了相應資料再加上自己的測試后，發現磁盤的順序讀寫速度和內存持平。

而且Linux對于磁盤的讀寫優化也比較多，包括read-ahead和write-behind，磁盤緩存等。如果在內存做這些操作的時候，一個是JAVA對象的內存開銷很大，另一個是隨著堆內存數據的增多，JAVA的GC時間會變得很長，使用磁盤操作有以下幾個好處：

• 磁盤緩存由Linux系統維護，減少了程序員的不少工作。
• 磁盤順序讀寫速度超過內存隨機讀寫。
• JVM的GC效率低，內存占用大。使用磁盤可以避免這一問題。
• 系統冷啟動后，磁盤緩存依然可用。

11、怎么解決kafka的數據丟失

producer端：

宏觀上看保證數據的可靠安全性，肯定是依據分區數做好數據備份，設立副本數。

broker端：

topic設置多分區，分區自適應所在機器，為了讓各分區均勻分布在所在的broker中，分區數要大于broker數。

分區是kafka進行并行讀寫的單位，是提升kafka速度的關鍵。

Consumer端

consumer端丟失消息的情形比較簡單：如果在消息處理完成前就提交了offset，那么就有可能造成數據的丟失。由于Kafka consumer默認是自動提交位移的，所以在后臺提交位移前一定要保證消息被正常處理了，因此不建議采用很重的處理邏輯，如果處理耗時很長，則建議把邏輯放到另一個線程中去做。為了避免數據丟失，現給出兩點建議：

• enable.auto.commit=false 關閉自動提交位移
• 在消息被完整處理之后再手動提交位移

12、fsimage和edit的區別?

• 大家都知道namenode與secondary namenode 的關系，當他們要進行數據同步時叫做checkpoint時就用到了fsimage與edit，fsimage是保存最新的元數據的信息，當fsimage數據到一定的大小事會去生成一個新的文件來保存元數據的信息，這個新的文件就是edit，edit會回滾最新的數據。

13、列舉幾個配置文件優化?

1)Core-site.xml 文件的優化

• a、fs.trash.interval，默認值： 0;說明： 這個是開啟hdfs文件刪除自動轉移到垃圾箱的選項，值為垃圾箱文件清除時間。一般開啟這個會比較好，以防錯誤刪除重要文件。單位是分鐘。
• b、dfs.namenode.handler.count，默認值：10;說明：hadoop系統里啟動的任務線程數，這里改為40，同樣可以嘗試該值大小對效率的影響變化進行最合適的值的設定。
• c、mapreduce.tasktracker.http.threads，默認值：40;說明：map和reduce是通過http進行數據傳輸的，這個是設置傳輸的并行線程數。

14、datanode 首次加入 cluster 的時候，如果 log 報告不兼容文件版本，那需要namenode 執行格式化操作，這樣處理的原因是?

• 1)這樣處理是不合理的，因為那么 namenode 格式化操作，是對文件系統進行格式化，namenode 格式化時清空 dfs/name 下空兩個目錄下的所有文件，之后，會在目錄 dfs.name.dir 下創建文件。
• 2)文本不兼容，有可能時 namenode 與 datanode 的 數據里的 namespaceID、clusterID 不一致，找到兩個 ID 位置，修改為一樣即可解決。

15、MapReduce 中排序發生在哪幾個階段?這些排序是否可以避免?為什么?

• 1)一個 MapReduce 作業由 Map 階段和 Reduce 階段兩部分組成，這兩階段會對數據排序，從這個意義上說，MapReduce 框架本質就是一個 Distributed Sort。
• 2)在 Map 階段，Map Task 會在本地磁盤輸出一個按照 key 排序(采用的是快速排序)的文件(中間可能產生多個文件，但最終會合并成一個)，在 Reduce 階段，每個 Reduce Task 會對收到的數據排序，這樣，數據便按照 Key 分成了若干組，之后以組為單位交給 reduce()處理。
• 3)很多人的誤解在 Map 階段，如果不使用 Combiner便不會排序，這是錯誤的，不管你用不用 Combiner，Map Task 均會對產生的數據排序(如果沒有 Reduce Task，則不會排序，實際上 Map 階段的排序就是為了減輕 Reduce端排序負載)。
• 4)由于這些排序是 MapReduce 自動完成的，用戶無法控制，因此，在hadoop 1.x 中無法避免，也不可以關閉，但 hadoop2.x 是可以關閉的。

16、hadoop的優化?

• 1)優化的思路可以從配置文件和系統以及代碼的設計思路來優化
• 2)配置文件的優化：調節適當的參數，在調參數時要進行測試
• 3)代碼的優化：combiner的個數盡量與reduce的個數相同，數據的類型保持一致，可以減少拆包與封包的進度
• 4)系統的優化：可以設置linux系統打開最大的文件數預計網絡的帶寬MTU的配置
• 5)為 job 添加一個 Combiner，可以大大的減少shuffer階段的maoTask拷貝過來給遠程的 reduce task的數據量，一般而言combiner與reduce相同。
• 6)在開發中盡量使用stringBuffer而不是string，string的模式是read-only的，如果對它進行修改，會產生臨時的對象，二stringBuffer是可修改的，不會產生臨時對象。
• 7)修改一下配置：以下是修改 mapred-site.xml 文件

a、修改最大槽位數：槽位數是在各個 tasktracker 上的 mapred-site.xml 上設置的，默認都是 2 

property  
namemapred.tasktracker.map.tasks.maximum/name  
value2/value  
/property  
property  
namemapred.tasktracker.reduce.tasks.maximum/name  
value2/value  
/property 

b、調整心跳間隔：集群規模小于 300 時，心跳間隔為 300 毫秒

• mapreduce.jobtracker.heartbeat.interval.min 心跳時間
• mapred.heartbeats.in.second 集群每增加多少節點，時間增加下面的值
• mapreduce.jobtracker.heartbeat.scaling.factor 集群每增加上面的個數，心跳增多少

c、啟動帶外心跳

• mapreduce.tasktracker.outofband.heartbeat 默認是 false

d、配置多塊磁盤

mapreduce.local.dir 

e、配置 RPC hander 數目

• mapred.job.tracker.handler.count 默認是 10，可以改成 50，根據機器的能力

f、配置 HTTP 線程數目

• tasktracker.http.threads 默認是 40，可以改成 100 根據機器的能力

g、選擇合適的壓縮方式，以 snappy 為例： 

property  
namemapred.compress.map.output/name  
valuetrue/value  
/property  
property  
namemapred.map.output.compression.codec/name  
valueorg.apache.hadoop.io.compress.SnappyCodec/value  
/property 

17、設計題

1)采集nginx產生的日志，日志的格式為user ip time url htmlId 每天產生的文件的數據量上億條，請設計方案把數據保存到HDFS上，并提供一下實時查詢的功能(響應時間小于3s)

A、某個用戶某天訪問某個URL的次數

B、某個URL某天被訪問的總次數

• 實時思路是：使用Logstash + Kafka + Spark-streaming + Redis + 報表展示平臺
• 離線的思路是：Logstash + Kafka + Elasticsearch + Spark-streaming + 關系型數據庫

A、B、數據在進入到Spark-streaming 中進行過濾，把符合要求的數據保存到Redis中

18、有 10 個文件，每個文件 1G，每個文件的每一行存放的都是用戶的 query，每個文件的query 都可能重復。要求你按照 query 的頻度排序。 還是典型的 TOP K 算法 ##，

解決方案如下：

1)方案 1：

順序讀取 10 個文件，按照 hash(query)%10 的結果將 query 寫入到另外 10 個文件(記為)中。這樣新生成的文件每個的大小大約也 1G(假設 hash 函數是隨機的)。 找一臺內存在 2G 左右的機器，依次對用 hash_map(query, query_count)來統計每個query 出現的次數。利用快速/堆/歸并排序按照出現次數進行排序。將排序好的 query 和對應的 query_cout 輸出到文件中。這樣得到了 10 個排好序的文件(記為)。 對這 10 個文件進行歸并排序(內排序與外排序相結合)。

2)方案 2：

一般 query 的總量是有限的，只是重復的次數比較多而已，可能對于所有的 query，一次性就可以加入到內存了。這樣，我們就可以采用 trie 樹/hash_map等直接來統計每個 query出現的次數，然后按出現次數做快速/堆/歸并排序就可以了。

3)方案 3：

與方案 1 類似，但在做完 hash，分成多個文件后，可以交給多個文件來處理，采用分布式的架構來處理(比如 MapReduce)，最后再進行合并。

19、在 2.5 億個整數中找出不重復的整數，注，內存不足以容納這 2.5 億個整數 。

• 方案 1：采用 2-Bitmap(每個數分配 2bit，00 表示不存在，01 表示出現一次，10 表示多次，11 無意義)進行，共需內存 2^32 * 2 bit=1 GB 內存，還可以接受。然后掃描這 2.5億個整數，查看 Bitmap 中相對應位，如果是 00 變 01，01 變 10，10 保持不變。所描完事后，查看 bitmap，把對應位是 01 的整數輸出即可。
• 方案 2：也可采用與第 1 題類似的方法，進行劃分小文件的方法。然后在小文件中找出不重復的整數，并排序。然后再進行歸并，注意去除重復的元素。

20、騰訊面試題：給 40 億個不重復的 unsigned int 的整數，沒排過序的，然后再給一個數，如何快速判斷這個數是否在那 40 億個數當中? ##

方案 1：oo，申請 512M 的內存，一個 bit 位代表一個 unsigned int 值。讀入 40 億個數，設置相應的 bit 位，讀入要查詢的數，查看相應 bit 位是否為 1，為 1 表示存在，為 0 表示不存在。

方案 2：這個問題在《編程珠璣》里有很好的描述，大家可以參考下面的思路，探討一下： 又因為 2^32 為 40 億多，所以給定一個數可能在，也可能不在其中; 這里我們把 40 億個數中的每一個用 32 位的二進制來表示 ，假設這 40 億個數開始放在一個文件中。 然后將這 40 億個數分成兩類:

• 1.最高位為 0
• 2.最高位為 1

并將這兩類分別寫入到兩個文件中，其中一個文件中數的個數=20 億，而另一個=20 億(這相當于折半了); 與要查找的數的最高位比較并接著進入相應的文件再查找 再然后把這個文件為又分成兩類:

• 1.次最高位為 0
• 2.次最高位為 1

并將這兩類分別寫入到兩個文件中，其中一個文件中數的個數=10 億，而另一個=10 億(這相當于折半了); 與要查找的數的次最高位比較并接著進入相應的文件再查找。

.....

以此類推，就可以找到了,而且時間復雜度為 O(logn)，方案 2 完。

3)附：這里，再簡單介紹下，位圖方法： 使用位圖法判斷整形數組是否存在重復 ,判斷集合中存在重復是常見編程任務之一，當集合中數據量比較大時我們通常希望少進行幾次掃描，這時雙重循環法就不可取了。

位圖法比較適合于這種情況，它的做法是按照集合中最大元素 max 創建一個長度為 max+1的新數組，然后再次掃描原數組，遇到幾就給新數組的第幾位置上 1，如遇到 5 就給新數組的第六個元素置 1，這樣下次再遇到 5 想置位時發現新數組的第六個元素已經是 1 了，這說明這次的數據肯定和以前的數據存在著重復。這 種給新數組初始化時置零其后置一的做法類似于位圖的處理方法故稱位圖法。它的運算次數最壞的情況為 2N。如果已知數組的最大值即能事先給新數組定長的話效 率還能提高一倍。

21、怎么在海量數據中找出重復次數最多的一個?

• 方案 1：先做 hash，然后求模映射為小文件，求出每個小文件中重復次數最多的一個，并記錄重復次數。然后找出上一步求出的數據中重復次數最多的一個就是所求(具體參考前面的題)。

22、上千萬或上億數據(有重復)，統計其中出現次數最多的錢 N 個數據。

• 方案 1：上千萬或上億的數據，現在的機器的內存應該能存下。所以考慮采用 hash_map/搜索二叉樹/紅黑樹等來進行統計次數。然后就是取出前 N 個出現次數最多的數據了，可以用第 2 題提到的堆機制完成。

23、一個文本文件，大約有一萬行，每行一個詞，要求統計出其中最頻繁出現的前 10 個詞，給出思想，給出時間復雜度分析 ##。

• 方案 1：這題是考慮時間效率。用 trie 樹統計每個詞出現的次數，時間復雜度是 O(nle)(le表示單詞的平準長度)。然后是找出出現最頻繁的前 10 個詞，可以用堆來實現，前面的題中已經講到了，時間復雜度是 O(nlg10)。所以總的時間復雜度，是 O(nle)與 O(nlg10)中較大的哪一 個。

24、100w 個數中找出最大的 100 個數 ##。

• 方案 1：在前面的題中，我們已經提到了，用一個含 100 個元素的最小堆完成。復雜度為O(100wlg100)。
• 方案 2：采用快速排序的思想，每次分割之后只考慮比軸大的一部分，知道比軸大的一部分在比 100 多的時候，采用傳統排序算法排序，取前 100 個。復雜度為 O(100w100)。
• 方案 3：采用局部淘汰法。選取前 100 個元素，并排序，記為序列 L。然后一次掃描剩余的元素 x，與排好序的 100 個元素中最小的元素比，如果比這個最小的 要大，那么把這個最小的元素刪除，并把 x 利用插入排序的思想，插入到序列 L 中。依次循環，直到掃描了所有的元素。復雜度為 O(100w*100)。

25、有一千萬條短信，有重復，以文本文件的形式保存，一行一條，有重復。 請用 5 分鐘時間，找出重復出現最多的前 10 條。

• 分析： 常規方法是先排序，在遍歷一次，找出重復最多的前 10 條。但是排序的算法復雜度最低為nlgn。
• 可以設計一個 hash_table, hash_mapstring, int，依次讀取一千萬條短信，加載到hash_table 表中，并且統計重復的次數，與此同時維護一張最多 10 條的短信表。 這樣遍歷一次就能找出最多的前 10 條，算法復雜度為 O(n)。