當前版本：

Ver 1.0

制作單位：

?

編寫人員：

審 核 人：

?

簽 收 人：

?

簽署日期：

??? 2017 年 05 月 22 日

版本號

1.0

版本日期

2017-05-22

所有者

?

作者

日期

描述

作者

版本號

2017-05-22

新增

?

?

?

?

?

?

?

?

?

?

?

?

?

?

?

?

?

?

?

?

?

?

?

?

?

?

?

?

?

?

?

?

?

?

?

?

?

分析：

hadoop 的集群是基于 master/slave 模式，namenode 和 jobtracker 屬于 master，datanode 和 tasktracker屬于 slave，master 只有一個，而 slave 有多個。SecondaryNameNode 內存需求和 NameNode 在一個數量級上，所以通常 secondary NameNode（運行在單獨的物理機器上）和 NameNode 運行在不同的機器上。

JobTracker 和 TaskTracker

JobTracker 對應于 NameNode

TaskTracker 對應于 DataNode

DataNode 和 NameNode 是針對數據存放來而言的

JobTracker 和 TaskTracker 是對于 MapReduce 執行而言的

mapreduce 中幾個主要概念，mapreduce 整體上可以分為這么幾條執行線索：

jobclient，JobTracker 與 TaskTracker。

1、JobClient 會在用戶端通過 JobClient 類將應用已經配置參數打包成 jar 文件存儲到 hdfs，并把路徑提交到 Jobtracker,然后由 JobTracker 創建每一個 Task（即 MapTask 和 ReduceTask）并將它們分發到各個 TaskTracker 服務中去執行

2、JobTracker 是一個 master 服務，軟件啟動之后 JobTracker 接收 Job，負責調度 Job 的每一個子任務 task運行于 TaskTracker 上，并監控它們，如果發現有失敗的 task 就重新運行它。一般情況應該把 JobTracker 部署在單獨的機器上。

3、TaskTracker 是運行在多個節點上的 slaver 服務。TaskTracker 主動與 JobTracker 通信，接收作業，并負責直接執行每一個任務。TaskTracker 都需要運行在 HDFS 的 DataNode 上

分析：

Client 向 NameNode 發起文件寫入的請求。

NameNode 根據文件大小和文件塊配置情況，返回給 Client 它所管理部分 DataNode 的信息。

Client 將文件劃分為多個 Block，根據 DataNode 的地址信息，按順序寫入到每一個 DataNode 塊中。

解析：

由于大數據面臨海量數據，讀寫數據都需要 io，然后還要冗余數據，hadoop 一般備 3 份數據，所以 IO就會打折扣。

分析：

A：puppetpuppet 是一種 Linux、Unix、windows 平臺的集中配置管理系統

B：pdsh 可以實現在在多臺機器上執行相同的命令

詳細參考：集群管理小工具介紹-pdsh

C：可以參考 Cloudera Manager 四大功能【翻譯】

首先這里給管理下一個定義：部署、配置、調試、監控，屬于管理

解析：

ganglia 作為一款最常用的 Linux 環境中的監控軟件，它擅長的的是從節點中按照用戶的需求以較低的代價

采集數據。但是 ganglia 在預警以及發生事件后通知用戶上并不擅長。最新的 ganglia 已經有了部分這方面

的功能。但是更擅長做警告的還有 Nagios。Nagios，就是一款精于預警、通知的軟件。通過將 Ganglia 和

Nagios 組合起來，把 Ganglia 采集的數據作為 Nagios 的數據源，然后利用 Nagios 來發送預警通知，可以

完美的實現一整套監控管理的系統。

分析：

Nagios 是集群監控工具，而且是云計算三大利器之一

分析：

SecondaryNameNode 是幫助恢復，而不是替代

分析：

第一套付費產品是 Cloudera Enterpris，Cloudera Enterprise 在美國加州舉行的 Hadoop 大會 (HadoopSummit) 上公開，以若干私有管理、監控、運作工具加強 Hadoop 的功能。收費采取合約訂購方式，價格隨用的 Hadoop 叢集大小變動。

分析：

NameNode 不需要從磁盤讀取 metadata，所有數據都在內存中，硬盤上的只是序列化的結果，只有每次namenode 啟動的時候才會讀取。

1）文件寫入

Client 向 NameNode 發起文件寫入的請求。

NameNode 根據文件大小和文件塊配置情況，返回給 Client 它所管理部分 DataNode 的信息。

Client 將文件劃分為多個 Block，根據 DataNode 的地址信息，按順序寫入到每一個 DataNode 塊中。

2）文件讀取

Client 向 NameNode 發起文件讀取的請求。

NameNode 返回文件存儲的 DataNode 的信息。

Client 讀取文件信息。

分析：

DataNode 是文件存儲的基本單元，它將 Block 存儲在本地文件系統中，保存了 Block 的 Meta-data，同時周期性地將所有存在的 Block 信息發送給 NameNode。

分析：

通過心跳機制。

（1）.長連接

Client 方與 Server 方先建立通訊連接，連接建立后不斷開，然后再進行報文發送和接收。這種方式下由于通訊連接一直存在，此種方式常用于點對點通訊。

（2）.短連接

Client 方與 Server 每進行一次報文收發交易時才進行通訊連接，交易完畢后立即斷開連接。此種方式常用于一點對多點通訊，比如多個 Client 連接一個 Server.

分析：

一旦 Slave 節點宕機，數據恢復是一個難題

分析：

hadoop dfsadmin -report

用這個命令可以快速定位出哪些節點 down 掉了，HDFS 的容量以及使用了多少，以及每個節點的硬盤使用情況。

當然 NameNode 有個 http 頁面也可以查詢，但是這個命令的輸出更適合我們的腳本監控 dfs 的使用狀況

分析：

首先明白什么是 RAID：磁盤陣列（Redundant Arrays of Independent Disks，RAID），有“獨立磁盤構成的具有冗余能力的陣列”之意。

這句話錯誤的地方在于太絕對，具體情況具體分析。題目不是重點，知識才是最重要的。

因為 hadoop 本身就具有冗余能力，所以如果不是很嚴格不需要都配備 RAID。

分析：

hadoop 為各個守護進程（namenode,secondarynamenode,jobtracker,datanode,tasktracker）統一分配的內存在 hadoop-env.sh 中設置，參數為 HADOOP_HEAPSIZE，默認為 1000M。

分析：

這個報錯是說明 DataNode 所裝的 Hadoop 版本和其它節點不一致，應該檢查 DataNode 的 Hadoop 版本

解析：

Hadoop 的基礎配置文件是 hadoop-default.xml，默認建立一個 Job 的時候會建立 Job 的 Config，Config

首先讀入hadoop-default.xml的配置，然后再讀入hadoop-site.xml的配置（這個文件初始的時候配置為空），

hadoop-site.xml 中主要配置需要覆蓋的 hadoop-default.xml 的系統級配置。具體配置可以參考下：

<property>

? <name>dfs.block.size</name>//block 的大小，單位字節，后面會提到用處，必須是 512 的倍數，因

為采用 crc 作文件完整性校驗，默認配置 512 是 checksum 的最小單元。

? <value>5120000</value>

? <description>The default block size for new files.</description>

</property>

分析：

lucene 是支持隨機讀寫的，而 hdfs 只支持隨機讀。但是 HBase 可以來補救。

HBase 提供隨機讀寫，來解決 Hadoop 不能處理的問題。HBase 自底層設計開始即聚焦于各種可伸縮性問題：表可以很―高‖，有數十億個數據行；也可以很―寬‖，有數百萬個列；水平分區并在上千個普通商用機節點上自動復制。表的模式是物理存儲的直接反映，使系統有可能提高高效的數據結構的序列化、存儲和檢索。

分析：

副本針對DataName而講的

分析：

支持c++等語言，需要通過接口。

分析：

一個task對應一個線程

分析：首先我們知道什么是 map 槽,map 槽->map slot，map slot 只是一個邏輯值 ( org.apache.hadoop.mapred.TaskTracker.TaskLauncher.numFreeSlots )，而不是對應著一個線程或者進程

分析：

? ?應該是一個block數組

1、運行mapred程序；
2、本次運行將生成一個Job，于是JobClient向JobTracker申請一個JobID以標識這個Job；
3、JobClient將Job所需要的資源提交到HDFS中一個以JobID命名的目錄中。這些資源包括JAR包、配置文件、InputSplit、等；
4、JobClient向JobTracker提交這個Job；
5、JobTracker初始化這個Job；
6、JobTracker從HDFS獲取這個Job的Split等信息；
7、JobTracker向TaskTracker分配任務；
8、TaskTracker從HDFS獲取這個Job的相關資源；
9、TaskTracker開啟一個新的JVM；
10、TaskTracker用新的JVM來執行Map或Reduce；

?

InputSplit也是一個interface，具體返回什么樣的implement，這是由具體的InputFormat來決定的。InputSplit也只有兩個接口函數：
long getLength() throws IOException;
String[] getLocations() throws IOException;
這個interface僅僅描述了Split有多長，以及存放這個Split的Location信息（也就是這個Split在HDFS上存放的機器。它可能有多個replication，存在于多臺機器上）。除此之外，就再沒有任何直接描述Split的信息了。比如：Split對應于哪個文件？在文件中的起始和結束位置是什么？等等重要的特征都沒有描述到。
為什么會這樣呢？因為關于Split的那些描述信息，對于MapReduce框架來說是不需要關心的。框架只關心Split的長度（主要用于一些統計信息）和Split的Location（主要用于Split的調度，后面會細說）。
而Split中真正重要的描述信息還是只有InputFormat會關心。在需要讀取一個Split的時候，其對應的InputSplit會被傳遞到InputFormat的第二個接口函數getRecordReader，然后被用于初始化一個RecordReader，以解析輸入數據。也就是說，描述Split的重要信息都被隱藏了，只有具體的InputFormat自己知道。它只需要保證getSplits返回的InputSplit和getRecordReader所關心的InputSplit是同樣的implement就行了。這就給InputFormat的實現提供了巨大的靈活性。

解答：

hdfs有namenode、secondraynamenode、datanode組成。

為n+1模式

namenode負責管理datanode和記錄元數據

secondraynamenode負責合并日志

datanode負責存儲數據

?

流程：

1.創建hadoop用戶

2.修改IP

3.安裝JDK，并配置環境變量

4.修改host文件映射

5.安裝SSH，配置無秘鑰通信

6.上傳解壓hadoop安裝包

7.配置conf文件夾下的hadoop-env.sh、core-site.xlmapre-site.xml、hdfs-site.xml

8.配置hadoop的環境變量

9.Hadoop namenode -format

10.start-all

解答：

1、權限問題，可能曾經用root啟動過集群。(例如hadoop搭建的集群,是tmp/hadoop-hadoop/.....)

2、可能是文件夾不存在

3、解決: 刪掉tmp下的那個文件,或改成當前用戶

解答：

1.namenode:管理集群，并記錄datanode文件信息。

2.Secondname:可以做冷備，對一定范圍內的數據做快照性備份。

3.Datanode：存儲數據。

4.Jobtracker：管理任務，并將任務分配給tasktracker。

5.Tasktracker:任務執行者

解答：

Hadoop的核心配置通過兩個xml文件來完成：

1.hadoop-default.xml；

2.hadoop-site.xml。

這些文件都使用xml格式，因此每個xml中都有一些屬性，包括名稱和值，但是當下這些文件都已不復存在。

解答：

Hadoop現在擁有3個配置文件：

1，core-site.xml；

2，hdfs-site.xml；

3，mapred-site.xml。

這些文件都保存在conf/子目錄下。

解答：

這個命令可以檢查Namenode、Datanode、Task Tracker、 Job Tracker是否正常工作。

解答：

解答：

error org.apache.hadoop.hdfs.server.namenode.NameNode

org.apache.hadoop.hdfs.server.common.inconsistentFSStateExceptio

n Directory /tmp/hadoop-root/dfs/name is in an inconsistent

state storage direction does not exist or is not accessible?

解答：

hadoop job -list 記錄job-id、hadoop job -kill job-id

?

hadoop fs -rmr /tmp/aaa

?

添加新節點：

hadoop -daemon.sh start datanode

hadoop -daemon.sh start tasktracker

?

移除一個節點：

hadoop mradmin -refreshnodes

hadoop dfsadmin -refreshnodes

解答：

1.FIFO schedular:默認，先進先出的原則

2.Capacity schedular:計算能力調度器，選擇占用最小，優先級高的先執行，以此類推。

3.Fair schedular:公平調度，所有的job具有相同的資源。

解答：

解答：

解答：

（1）? 從應用程序角度進行優化。由于mapreduce是迭代逐行解析數據文件的，怎樣在迭代的情況下，編寫高效率的應用程序，是一種優化思路。

（2）? 對Hadoop參數進行調優。當前hadoop系統有190多個配置參數，怎樣調整這些參數，使hadoop作業運行盡可能的快，也是一種優化思路。

（3） 從系統實現角度進行優化。這種優化難度是最大的，它是從hadoop實現機制角度，發現當前Hadoop設計和實現上的缺點，然后進行源碼級地修改。該方法雖難度大，但往往效果明顯。

（4）linux內核參數調整

1. 使用自定義Writable

自帶的Text很好用，但是字符串轉換開銷較大，故根據實際需要自定義Writable，注意作為Key時要實現WritableCompareable接口

避免output.collect(new Text( ),new Text())

提倡key.set( ) value.set( ) output.collect(key,value)

前者會產生大量的Text對象，使用完后Java垃圾回收器會花費大量的時間去收集這些對象

?

2. 使用StringBuilder

不要使用Formatter StringBuffer（?線程安全）

StringBuffer盡量少使用多個append方法，適當使用+

?

3. 使用DistributedCache加載文件

比如配置文件，詞典，共享文件，避免使用static變量

?

4. 充分使用Combiner Parttitioner Comparator。

Combiner : 對map任務進行本地聚合

Parttitioner ： 合適的Parttitioner避免reduce端負載不均

Comparator ： 二次排序

比如求每天的最大氣溫，map結果為日期：氣溫，若氣溫是降序的，直接取列表首元素即可

?

5. 使用自定義InputFormat和OutputFormat

?

6. MR應避免

?

靜態變量：不能用于計數，應使用Counter

?

大對象：Map List

?

遞歸：避免遞歸深度過大

?

超長正則表達式：消耗性能，要在map或reduce函數外編譯正則表達式

?

不要創建本地文件：變向的把HDFS里面的數據轉移到TaskTracker，占用網絡帶寬

?

不要大量創建目錄和文件

?

不要大量使用System.out.println，而使用Logger

?

不要自定義過多的Counter，最好不要超過100個

?

不要配置過大內存，mapred.child.java.opts -Xmx2000m是用來設置mapreduce任務使用的最大heap量

?

7.關于map的數目

map數目過大[創建和初始化map的開銷]，一般是由大量小文件造成的，或者dfs.block.size設置的太小，對于小文件可以archive文件或者Hadoop fs -merge合并成一個大文件.

map數目過少，造成單個map任務執行時間過長，頻繁推測執行，且容易內存溢出，并行性優勢不能體現出來。dfs.block.size一般為256M-512M

壓縮的Text 文件是不能被分割的，所以盡量使用SequenceFile，可以切分

?

8.關于reduce的數目

reduce數目過大，產生大量的小文件，消耗大量不必要的資源，reduce數目過低呢，造成數據傾斜問題，且通常不能通過修改參數改變。

可選方案：mapred.reduce.tasks設為-1變成AutoReduce。

Key的分布，也在某種程度上決定了Reduce數目，所以要根據Key的特點設計相對應的Parttitioner 避免數據傾斜

?

9.Map-side相關參數優化

io.sort.mb（100MB）：通常k個map tasks會對應一個buffer，buffer主要用來緩存map部分計算結果，并做一些預排序提高map性能，若map輸出結果較大，可以調高這個參數，減少map任務進行spill任務個數，降低 I/O的操作次數。若map任務的瓶頸在I/O的話，那么將會大大提高map性能。如何判斷map任務的瓶頸？

io.sort.spill.percent(0.8)：spill操作就是當內存buffer超過一定閾值（這里通常是百分比）的時候，會將buffer中得數據寫到Disk中。而不是等buffer滿后在spill，否則會造成map的計算任務等待buffer的釋放。一般來說，調整 io.sort.mb而不是這個參數。

io.sort.factor（10）：map任務會產生很多的spill文件，而map任務在正常退出之前會將這些spill文件合并成一個文件，即merger過程，缺省是一次合并10個參數，調大io.sort.factor，減少merge的次數，減少Disk I/O操作，提高map性能。

min.num.spill.for.combine：通常為了減少map和reduce數據傳輸量，我們會制定一個combiner，將map結果進行本地聚集。這里combiner可能在merger之前，也可能在其之后。那么什么時候在其之前呢？當spill個數至少為min.num.spill.for.combine指定的數目時同時程序指定了Combiner，Combiner會在其之前運行，減少寫入到Disk的數據量，減少I/O次數。

?

10.壓縮（時間換空間）

MR中的數據無論是中間數據還是輸入輸出結果都是巨大的，若不使用壓縮不僅浪費磁盤空間且會消耗大量網絡帶寬。同樣在spill，merge（reduce也對有一個merge）亦可以使用壓縮。若想在cpu時間和壓縮比之間尋找一個平衡，LzoCodec比較適合。通常MR任務的瓶頸不在CPU而在于I/O，所以大部分的MR任務都適合使用壓縮。

?

11. reduce-side相關參數優化

reduce：copy->sort->reduce，也稱shuffle

mapred.reduce.parellel.copies（5）：任一個map任務可能包含一個或者多個reduce所需要數據，故一個map任務完成后，相應的reduce就會立即啟動線程下載自己所需要的數據。調大這個參數比較適合map任務比較多且完成時間比較短的Job。

mapred.reduce.copy.backoff：reduce端從map端下載數據也有可能由于網絡故障，map端機器故障而失敗。那么reduce下載線程肯定不會無限等待，當等待時間超過mapred.reduce.copy.backoff時，便放棄，嘗試從其他地方下載。需注意：在網絡情況比較差的環境，我們需要調大這個參數，避免reduce下載線程被誤判為失敗。

io.sort.factor：recude將map結果下載到本地時，亦需要merge，如果reduce的瓶頸在于I/O，可嘗試調高增加merge的并發吞吐，提高reduce性能、

mapred.job.shuffle.input.buffer.percent（0.7）：reduce從map下載的數據不會立刻就寫到Disk中，而是先緩存在內存中，mapred.job.shuffle.input.buffer.percent指定內存的多少比例用于緩存數據，內存大小可通過mapred.child.java.opts來設置。和map類似，buffer不是等到寫滿才往磁盤中寫，也是到達閾值就寫，閾值由mapred.job,shuffle.merge.percent來指定。若Reduce下載速度很快，容易內存溢出，適當增大這個參數對增加reduce性能有些幫助。

mapred.job.reduce.input.buffer.percent (0)：當Reduce下載map數據完成之后，就會開始真正的reduce的計算，reduce的計算必然也是要消耗內存的，那么在讀物reduce所需要的數據時，同樣需要內存作為buffer，這個參數是決定多少的內存百分比作為buffer。默認為0，也就是說reduce全部從磁盤讀數據。若redcue計算任務消耗內存很小，那么可以設置這個參數大于0，使一部分內存用來緩存數據。

?

解答：

（1） 避免不必要的reduce任務

如果mapreduce程序中reduce是不必要的，那么我們可以在map中處理數據, Reducer設置為0。這樣避免了多余的reduce任務。

（2） 為job添加一個Combiner

為job添加一個combiner可以大大減少shuffle階段從map task拷貝給遠程reduce task的數據量。一般而言，combiner與reducer相同。

（3） 根據處理數據特征使用最適合和簡潔的Writable類型

Text對象使用起來很方便，但它在由數值轉換到文本或是由UTF8字符串轉換到文本時都是低效的，且會消耗大量的CPU時間。當處理那些非文本的數據時，可以使用二進制的Writable類型，如IntWritable， FloatWritable等。二進制writable好處：避免文件轉換的消耗；使map task中間結果占用更少的空間。

（4） 重用Writable類型

很多MapReduce用戶常犯的一個錯誤是，在一個map/reduce方法中為每個輸出都創建Writable對象。例如，你的Wordcout mapper方法可能這樣寫：

?

public void map(...) {

? …

?

? for (String word : words) {

??? output.collect(new Text(word), new IntWritable(1));

? }

}

這樣會導致程序分配出成千上萬個短周期的對象。Java垃圾收集器就要為此做很多的工作。更有效的寫法是：

class MyMapper … {

? Text wordText = new Text();

? IntWritable one = new IntWritable(1);

? public void map(...) {

??? for (String word: words) {

????? wordText.set(word);

????? output.collect(wordText, one);

??? }

? }

}

（5） 使用StringBuffer而不是String

當需要對字符串進行操作時，使用StringBuffer而不是String，String是read-only的，如果對它進行修改，會產生臨時對象，而StringBuffer是可修改的，不會產生臨時對象。

解答：

當分備份數為1時。

解答：

出現在map階段的map方法后。

解答：

這個datanode的數據會在其他的datanode上重新做備份。

解答：

?

解答：

?

解答：

先分析宕機后的損失，宕機后直接導致client無法訪問，內存中的元數據丟失，但是硬盤中的元數據應該還存在，如果只是節點掛了，重啟即可，如果是機器掛了，重啟機器后看節點是否能重啟，不能重啟就要找到原因修復了。但是最終的解決方案應該是在設計集群的初期就考慮到這個問題，做namenode的HA。

?

解答：

Datanode宕機了后，如果是短暫的宕機，可以實現寫好腳本監控，將它啟動起來。如果是長時間宕機了，那么datanode上的數據應該已經被備份到其他機器了，那這臺datanode就是一臺新的datanode了，刪除他的所有數據文件和狀態文件，重新啟動。

解答：

在Reduce階段，先對Key排序，再對Value排序

最常用的方法是將Value放到Key中，實現一個組合Key，然后自定義Key排序規則（為Key實現一個WritableComparable）。

解答：

第一種情況比較簡單，只需將小表放到DistributedCache中即可；

第二種情況常用的方法有：map-side join（要求輸入數據有序，通常用戶Hbase中的數據表連接），reduce-side join，semi join（半連接）

解答：

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

解答：

combiner是reduce的實現，在map端運行計算任務，減少map端的輸出數據。

作用就是優化。

但是combiner的使用場景是mapreduce的map輸出結果和reduce輸入輸出一樣。

?

partition的默認實現是hashpartition，是map端將數據按照reduce個數取余，進行分區，不同的reduce來copy自己的數據。

partition的作用是將數據分到不同的reduce進行計算，加快計算效果。

?

?

1、combiner最基本是實現本地key的聚合，對map輸出的key排序，value進行迭代。如下所示：

　　map: (K1, V1) → list(K2, V2)

　　combine: (K2, list(V2)) → list(K2, V2)

　　reduce: (K2, list(V2)) → list(K3, V3)

　　2、combiner還具有類似本地的reduce功能.

　　例如hadoop自帶的wordcount的例子和找出value的最大值的程序，combiner和reduce完全一致。如下所示：

　　map: (K1, V1) → list(K2, V2)

　　combine: (K2, list(V2)) → list(K3, V3)

　　reduce: (K3, list(V3)) → list(K4, V4)

　　3、如果不用combiner，那么，所有的結果都是reduce完成，效率會相對低下。使用combiner，先完成的map會在本地聚合，提升速度。

　　4、對于hadoop自帶的wordcount的例子，value就是一個疊加的數字，所以map一結束就可以進行reduce的value疊加，而不必要等到所有的map結束再去進行reduce的value疊加。

　　combiner使用的合適，可以在滿足業務的情況下提升job的速度，如果不合適，則將導致輸出的結果不正確。

?

解答：

解答：

分別說明hdfs，yarn，mapreduce

解答：

解答：

解答：

解答：

解答：

解答：

Map—combiner—partition—sort—copy—sort—grouping—reduce

解答：

解答：

解答：

解答：

解答：

解答：

解答：

解答：

解答：

數據傾斜：map /reduce程序執行時，reduce節點大部分執行完畢，但是有一個或者幾個reduce節點運行很慢，導致整個程序的處理時間很長，這是因為某一個key的條數比其他key多很多（有時是百倍或者千倍之多），這條key所在的reduce節點所處理的數據量比其他節點就大很多，從而導致某幾個節點遲遲運行不完，此稱之為數據傾斜。

?

用hadoop程序進行數據關聯時，常碰到數據傾斜的情況，這里提供一種解決方法。

自己實現partition類，用key和value相加取hash值：

方式1：

源代碼：

public int getPartition(K key, V value,

????????????????????????? int numReduceTasks) {

??? return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;

? }

修改后

public int getPartition(K key, V value,

????????????????????????? int numReduceTasks) {

??? return ((（key).hashCode()+value.hashCode()） & Integer.MAX_VALUE) % numReduceTasks;

? }

?

方式2：

public class HashPartitioner<K, V> extends Partitioner<K, V> {

private int aa= 0;

? /** Use {@link Object#hashCode()} to partition. */

? public int getPartition(K key, V value,

????????????????????????? int numReduceTasks) {

??? return (key.hashCode()+(aa++) & Integer.MAX_VALUE) % numReduceTasks;

? }

?

解答：

解答：

解答：

解答：

combiner是reduce的實現，在map端運行計算任務，減少map端的輸出數據。

作用就是優化。

但是combiner的使用場景是mapreduce的map和reduce輸入輸出一樣。

解答：

map的數量有數據塊決定，reduce數量隨便配置。

解答：

（1）它減少了存儲文件所需的空間；

（2）加快了數據在網絡上或者從磁盤上或到磁盤上的傳輸速度；

解答：

一個MapReduce作業的生命周期大體分為5個階段?【1】?：

1.?作業提交與初始化

2.?任務調度與監控

3. 任務運行環境準備

4. 任務執行

5. 作業完成

我們假設JobTracker已經啟動，那么調度器是怎么啟動的？JobTracker在啟動時有以下代碼：

JobTracker tracker = startTracker(new JobConf());

tracker.offerService();

其中offerService方法負責啟動JobTracker提供的各個服務，有這樣一行代碼：

taskScheduler.start();

taskScheduler即為任務調度器。start方法是抽象類TaskScheduler提供的接口，用于啟動調度器。每個調度器類都要繼承TaskScheduler類。回憶一下，調度器啟動時會將各個監聽器對象注冊到JobTracker，以FIFO調度器JobQueueTaskScheduler為例：

@Override

? public synchronized void start() throws IOException {

??? super.start();

??? taskTrackerManager.addJobInProgressListener(jobQueueJobInProgressListener);

??? eagerTaskInitializationListener.setTaskTrackerManager(taskTrackerManager);

??? eagerTaskInitializationListener.start();

??? taskTrackerManager.addJobInProgressListener(

??????? eagerTaskInitializationListener);

? }

這里注冊了兩個監聽器，其中eagerTaskInitializationListener負責作業初始化，而jobQueueJobInProgressListener則負責作業的執行和監控。當有作業提交到JobTracker時，JobTracker會執行所有訂閱它消息的監聽器的jobAdded方法。對于eagerTaskInitializationListener來說：?

@Override

? public void jobAdded(JobInProgress job) {

??? synchronized (jobInitQueue) {

????? jobInitQueue.add(job);

????? resortInitQueue();

????? jobInitQueue.notifyAll();

??? }

? }

提交的作業的JobInProgress對象被添加到作業初始化隊列jobInitQueue中，并喚醒初始化線程（若原來沒有作業可以初始化）：

class JobInitManager implements Runnable {

??? public void run() {

????? JobInProgress job = null;

????? while (true) {

??????? try {

????????? synchronized (jobInitQueue) {

??????????? while (jobInitQueue.isEmpty()) {

????????????? jobInitQueue.wait();

??????????? }

??????????? job = jobInitQueue.remove(0);

????????? }

????????? threadPool.execute(new InitJob(job));

??????? } catch (InterruptedException t) {

????????? LOG.info("JobInitManagerThread interrupted.");

????????? break;

??????? }

????? }

????? threadPool.shutdownNow();

??? }

? }

這種工作方式是一種“生產者-消費者”模式：作業初始化線程是消費者，而監聽器eagerTaskInitializationListener是生產者。這里可以有多個消費者線程，放到一個固定資源的線程池中，線程個數通過mapred.jobinit.threads參數配置，默認為4個。

下面我們重點來看調度器中的另一個監聽器。?jobQueueJobInProgressListener對象在調度器中初始化時連續執行了兩個構造器完成初始化：

public JobQueueJobInProgressListener() {

??? this(new TreeMap<JobSchedulingInfo,

???????????????????? JobInProgress>(FIFO_JOB_QUEUE_COMPARATOR));

? }

? /**

?? * For clients that want to provide their own job priorities.

?? * @param jobQueue A collection whose iterator returns jobs in priority order.

?? */

? protected JobQueueJobInProgressListener(Map<JobSchedulingInfo,

????????????????????????????????????????? JobInProgress> jobQueue) {

??? this.jobQueue = Collections.synchronizedMap(jobQueue);

? }

其中，第一個構造器調用重載的第二個構造器。可以看到，調度器使用一個隊列jobQueue來保存提交的作業。這個隊列使用一個TreeMap對象實現，TreeMap的特點是底層使用紅黑樹實現，可以按照鍵來排序，并且由于是平衡樹，效率較高。作為鍵的是一個JobSchedulingInfo對象，作為值就是提交的作業對應的JobInProgress對象。另外，由于TreeMap本身不是線程安全的，這里使用了集合類的同步方法構造了一個線程安全的Map。使用帶有排序功能的數據結構的目的是使作業在隊列中按照優先級的大小排列，這樣每次調度器只需從隊列頭部獲得作業即可。

作業的順序由優先級決定，而優先級信息包含在JobSchedulingInfo對象中：

static class JobSchedulingInfo {

??? private JobPriority priority;

??? private long startTime;

??? private JobID id;

...

}

該對象包含了作業的優先級、ID和開始時間等信息。在Hadoop中，作業的優先級有以下五種：VERY_HIGH、HIGH、NORMAL、LOW、VERY_LOW。這些字段是通過作業的JobStatus對象初始化的。由于該對象作為TreeMap的鍵，因此要實現自己的equals方法和hashCode方法：

@Override

??? public boolean equals(Object obj) {

????? if (obj == null || obj.getClass() != JobSchedulingInfo.class) {

??????? return false;

????? } else if (obj == this) {

??????? return true;

????? }

????? else if (obj instanceof JobSchedulingInfo) {

??????? JobSchedulingInfo that = (JobSchedulingInfo)obj;

??????? return (this.id.equals(that.id) &&

??????????????? this.startTime == that.startTime &&

??????????????? this.priority == that.priority);

????? }

????? return false;

??? }

我們看到，兩個JobSchedulingInfo對象相等的條件是類型一致，并且作業ID、開始時間和優先級都相等。hashCode的計算比較簡單：

@Override

??? public int hashCode() {

????? return (int)(id.hashCode() * priority.hashCode() + startTime);

??? }

注意，監聽器的第一個構造器有一個比較器參數，用于定義?JobSchedulingInfo的比較方式：

static final Comparator<JobSchedulingInfo> FIFO_JOB_QUEUE_COMPARATOR

??? = new Comparator<JobSchedulingInfo>() {

??? public int compare(JobSchedulingInfo o1, JobSchedulingInfo o2) {

????? int res = o1.getPriority().compareTo(o2.getPriority());

???? ?if (res == 0) {

??????? if (o1.getStartTime() < o2.getStartTime()) {

????????? res = -1;

??????? } else {

????????? res = (o1.getStartTime() == o2.getStartTime() ? 0 : 1);

?????? }

????? }

????? if (res == 0) {

??????? res = o1.getJobID().compareTo(o2.getJobID());

????? }

????? return res;

??? }

? };

從上面看出，首先比較作業的優先級，若優先級相等則比較開始時間（FIFO），若再相等則比較作業ID。?我們在實現自己的調度器時可能要定義自己的作業隊列，那么作業在隊列中的順序（即?JobSchedulingInfo的比較器?）就要仔細定義，這是調度器能夠正常運行基礎。

Hadoop中的作業調度采用pull方式，即TaskTracker定時向JobTracker發送心跳信息索取一個新的任務，這些信息包括數據結點上作業和任務的運行情況，以及該TaskTracker上的資源使用情況。JobTracker會依據以上信息更新作業隊列的狀態，并調用調度器選擇一個或多個任務以心跳響應的形式返回給TaskTracker。從上面描述可以看出，JobTracker和taskScheduler之間的互相利用關系：前者利用后者為TaskTracker分配任務；后者利用前者更新隊列和作業信息。接下來，我們一步步詳述該過程。

首先，當一個心跳到達JobTracker時（實際上這是一個來自TaskTracker的遠程過程調用?heartbeat方法?，協議接口是InterTrackerProtocol），會執行兩種動作：更新狀態和下達命令?【1】?。下達命令稍后關注。有關更新狀態的一些代碼片段如下：

if (!processHeartbeat(status, initialContact, now)) {

????? if (prevHeartbeatResponse != null) {

??????? trackerToHeartbeatResponseMap.remove(trackerName);

????? }

????? return new HeartbeatResponse(newResponseId,

?????????????????? new TaskTrackerAction[] {new ReinitTrackerAction()});

??? }

具體的心跳處理，由私有函數processHeartbeat完成。該函數中有以下兩個方法調用：

updateTaskStatuses(trackerStatus);

??? updateNodeHealthStatus(trackerStatus, timeStamp);

分別用來更新任務的狀態和結點的健康狀態。在第一個方法中有下面代碼片段：

TaskInProgress tip = taskidToTIPMap.get(taskId);

????? // Check if the tip is known to the jobtracker. In case of a restarted

????? // jt, some tasks might join in later

????? if (tip != null || hasRestarted()) {

??????? if (tip == null) {

????????? tip = job.getTaskInProgress(taskId.getTaskID());

????????? job.addRunningTaskToTIP(tip, taskId, status, false);

??????? }

???????

??????? // Update the job and inform the listeners if necessary

??????? JobStatus prevStatus = (JobStatus)job.getStatus().clone();

??????? // Clone TaskStatus object here, because JobInProgress

??????? // or TaskInProgress can modify this object and

??????? // the changes should not get reflected in TaskTrackerStatus.

??????? // An old TaskTrackerStatus is used later in countMapTasks, etc.

??????? job.updateTaskStatus(tip, (TaskStatus)report.clone());

??????? JobStatus newStatus = (JobStatus)job.getStatus().clone();

??????? // Update the listeners if an incomplete job completes

??????? if (prevStatus.getRunState() != newStatus.getRunState()) {

????????? JobStatusChangeEvent event =

??????????? new JobStatusChangeEvent(job, EventType.RUN_STATE_CHANGED,

???????????????????????????????????? prevStatus, newStatus);

????????? updateJobInProgressListeners(event);

??????? }

????? } else {

??????? LOG.info("Serious problem.? While updating status, cannot find taskid "

???????????????? + report.getTaskID());

????? }

這里的job對象通過從TaskTracker那里得到的task狀態信息中抽取出來。注意，這里拷貝了原有作業狀態的一個副本，然后修改這個副本的相關信息，調用的是updateJobStatus方法，更新任務的狀態信息和JobInProgress的相關信息，如map和reduce任務的進度等，這里不展開了。這些信息的更新可以為調度器的工作提供依據。

作業狀態的更新是通過updateJobInProgressListeners方法實現，該方法的參數是一個JobStatusChangeEvent對象，表示作業狀態變化的事件。這種事件的類型可以是運行狀態改變、開始時間改變、優先級改變等等。用戶也可以根據需要自定義事件類型。事件對象維護了兩個JobStatus對象，分別表示事件發生前后作業的狀態。?
進入該方法后，我們又看到了熟悉的觀察者模式：

// Update the listeners about the job

? // Assuming JobTracker is locked on entry.

? private void updateJobInProgressListeners(JobChangeEvent event) {

??? for (JobInProgressListener listener : jobInProgressListeners) {

????? listener.jobUpdated(event);

??? }

? }

這次每個監聽器要回調jobUpdated方法，表示作業有更新。對于jobQueueJobInProgressListener來說是這樣做的：

@Override

? public synchronized void jobUpdated(JobChangeEvent event) {

??? JobInProgress job = event.getJobInProgress();

??? if (event instanceof JobStatusChangeEvent) {

????? // Check if the ordering of the job has changed

????? // For now priority and start-time can change the job ordering

????? JobStatusChangeEvent statusEvent = (JobStatusChangeEvent)event;

????? JobSchedulingInfo oldInfo =?

??????? new JobSchedulingInfo(statusEvent.getOldStatus());

????? if (statusEvent.getEventType() == EventType.PRIORITY_CHANGED

????????? || statusEvent.getEventType() == EventType.START_TIME_CHANGED) {

??????? // Make a priority change

??????? reorderJobs(job, oldInfo);

????? } else if (statusEvent.getEventType() == EventType.RUN_STATE_CHANGED) {

??????? // Check if the job is complete

??????? int runState = statusEvent.getNewStatus().getRunState();

??????? if (runState == JobStatus.SUCCEEDED

??????????? || runState == JobStatus.FAILED

??????????? || runState == JobStatus.KILLED) {

??? ??????jobCompleted(oldInfo);

??????? }

????? }

??? }

? }

首先，獲取作業更新?前?的狀態。然后根據事件的類型，進行相應的處理。比如，如果優先級變化了，則要重新排列隊列中作業的順序。這里直接取出原有作業，重新插入隊列。插入后，作業會自動重新排序，體現了TreeMap的優越性。再比如，如果作業狀態變為完成，那么就從隊列中刪除該作業。

private void reorderJobs(JobInProgress job, JobSchedulingInfo oldInfo) {

??? synchronized (jobQueue) {

????? jobQueue.remove(oldInfo);

????? jobQueue.put(new JobSchedulingInfo(job), job);

??? }

? }

下面就是調度器中最關鍵的一步了：任務選擇。此時，作業隊列中信息已經更新完畢，可以選擇一些任務返回給TaskTracker執行了。heartbeat方法接下來會有這樣的代碼：

List<Task> tasks = getSetupAndCleanupTasks(taskTrackerStatus);

? if (tasks == null ) {

??? tasks = taskScheduler.assignTasks(taskTrackers.get(trackerName));

? }

如果不需要setup和cleanup，就說明需要選擇map或reduce任務。調用TaskScheduler的assignTasks方法完成任務選擇。由于篇幅限制，我打算將這部分內容放到下一篇文章中，并關注heartbeat中JobTracker下達的命令過程以及JobInProgress和TaskInProgress對調度有影響的一些字段。

?

流程：

1.client鏈接namenode存數據

2.namenode記錄一條數據位置信息（元數據），告訴client存哪。

3.client用hdfs的api將數據塊（默認是64M）存儲到datanode上。

4.datanode將數據水平備份。并且備份完將反饋client。

5.client通知namenode存儲塊完畢。

6.namenode將元數據同步到內存中。

7.另一塊循環上面的過程。

?

流程：

1.client鏈接namenode，查看元數據，找到數據的存儲位置。

2.client通過hdfs的api并發讀取數據。

3.關閉連接。

?

解答：

?

??? Word count例子接口

============================

一個MapReduce作業（job）通常會把輸入的數據集切分為若干獨立的數據塊，由map任務（task）以完全并行的方式處理它們。框架會對map的輸出先進行排序，然后把結果輸入給reduce任務。通常作業的輸入和輸出都會被存儲在文件系統中。整個框架負責任務的調度和監控，以及重新執行已經失敗的任務。

　　通常，MapReduce框架和分布式文件系統是運行在一組相同的節點上的，也就是說，計算節點和存儲節點通常在一起。這種配置允許框架在那些已經存好數據的節點上高效地調度任務，這可以使整個集群的網絡帶寬被非常高效地利用。

　　MapReduce框架由一個單獨的master JobTracker和每個集群節點一個slave TaskTracker共同組成。master負責調度構成一個作業的所有任務，這些任務分布在不同的slave上，master監控它們的執行，重新執行已經失敗的任務。而slave僅負責執行由master指派的任務

?

?

解答：

1)????? 用戶向YARN 中提交應用程序， 其中包括ApplicationMaster 程序、啟動ApplicationMaster 的命令、用戶程序等。

2)????? ResourceManager 為該應用程序分配第一個Container， 并與對應的NodeManager 通信，要求它在這個Container 中啟動應用程序的ApplicationMaster。

3)????? ApplicationMaster 首先向ResourceManager 注冊， 這樣用戶可以直接通過ResourceManage 查看應用程序的運行狀態，然后它將為各個任務申請資源，并監控它的運行狀態，直到運行結束，即重復步驟4~7。

4)????? ApplicationMaster 采用輪詢的方式通過RPC 協議向ResourceManager 申請和領取資源。

5)????? 一旦ApplicationMaster 申請到資源后，便與對應的NodeManager 通信，要求它啟動任務。

6)????? NodeManager 為任務設置好運行環境（包括環境變量、JAR 包、二進制程序等）后，將任務啟動命令寫到一個腳本中，并通過運行該腳本啟動任務。

7)????? 各個任務通過某個RPC 協議向ApplicationMaster 匯報自己的狀態和進度，以讓ApplicationMaster 隨時掌握各個任務的運行狀態，從而可以在任務失敗時重新啟動任務。在應用程序運行過程中，用戶可隨時通過RPC 向ApplicationMaster 查詢應用程序的當前運行狀態。

8)????? 應用程序運行完成后，ApplicationMaster 向ResourceManager 注銷并關閉自己。

?

解答：

1、內存數據庫derby，安裝小，但是數據存在內存，不穩定

2、mysql數據庫，數據存儲模式可以自己設置，持久化好，查看方便。

解答：

內部表：加載數據到hive所在的hdfs目錄，刪除時，元數據和數據文件都刪除

外部表：不加載數據到hive所在的hdfs目錄，刪除時，只刪除表結構。

解答：

1、因為外部表不會加載數據到hive，減少數據傳輸、數據還能共享。

2、hive不會修改數據，所以無需擔心數據的損壞

3、刪除表時，只刪除表結構、不刪除數據。

解答：

在導入hive的時候，如果數據庫中有blob或者text字段，會報錯。有個參數limit

解答：

Hive提供了三個虛擬列：

INPUT__FILE__NAME

BLOCK__OFFSET__INSIDE__FILE

ROW__OFFSET__INSIDE__BLOCK

但ROW__OFFSET__INSIDE__BLOCK默認是不可用的，需要設置hive.exec.rowoffset為true才可以。可以用來排查有問題的輸入數據。

INPUT__FILE__NAME, mapper任務的輸出文件名。

BLOCK__OFFSET__INSIDE__FILE, 當前全局文件的偏移量。對于塊壓縮文件，就是當前塊的文件偏移量，即當前塊的第一個字節在文件中的偏移量。

hive> SELECT INPUT__FILE__NAME, BLOCK__OFFSET__INSIDE__FILE, line

?

> FROM hive_text WHERE line LIKE '%hive%' LIMIT 2;

?

har://file/user/hive/warehouse/hive_text/folder=docs/

?

data.har/user/hive/warehouse/hive_text/folder=docs/README.txt? 2243

?

har://file/user/hive/warehouse/hive_text/folder=docs/

?

data.har/user/hive/warehouse/hive_text/folder=docs/README.txt? 3646

?

解答：

分區表，動態分區

解答：

insert?into：將某一張表中的數據寫到另一張表中

override?write：覆蓋之前的內容。

解答：

alter table ptable drop partition (daytime='20140911',city='bj');

元數據，數據文件都刪除，但目錄daytime= 20140911還在

?

解答：

提示：Hive中的left semi join替換sql中的in操作

解答：

解答：

解答：

hbase存儲時，數據按照Row key的字典序(byte order)排序存儲。設計key時，要充分排序存儲這個特性，將經常一起讀取的行存儲放到一起。(位置相關性)

一個列族在數據底層是一個文件，所以將經常一起查詢的列放到一個列族中，列族盡量少，減少文件的尋址時間。

因為hbase是列式數據庫，列非表schema的一部分，所以在設計初期只需要考慮rowkey 和 columnFamily即可，rowkey有位置相關性，所以如果數據是練習查詢的，最好對同類數據加一個前綴，而每個columnFamily實際上在底層是一個文件，那么文件越小，查詢越快，所以講經常一起查詢的列設計到一個列簇，但是列簇不宜過多。

?

?Rowkey長度原則

Rowkey是一個二進制碼流，Rowkey的長度被很多開發者建議說設計在10~100個字節，不過建議是越短越好，不要超過16個字節。

原因如下：

（1）數據的持久化文件HFile中是按照KeyValue存儲的，如果Rowkey過長比如100個字節，1000萬列數據光Rowkey就要占用100*1000萬=10億個字節，將近1G數據，這會極大影響HFile的存儲效率；

（2）MemStore將緩存部分數據到內存，如果Rowkey字段過長內存的有效利用率會降低，系統將無法緩存更多的數據，這會降低檢索效率。因此Rowkey的字節長度越短越好。

（3）目前操作系統是都是64位系統，內存8字節對齊。控制在16個字節，8字節的整數倍利用操作系統的最佳特性。

Rowkey散列原則

如果Rowkey是按時間戳的方式遞增，不要將時間放在二進制碼的前面，建議將Rowkey的高位作為散列字段，由程序循環生成，低位放時間字段，這樣將提高數據均衡分布在每個Regionserver實現負載均衡的幾率。如果沒有散列字段，首字段直接是時間信息將產生所有新數據都在一個 RegionServer上堆積的熱點現象，這樣在做數據檢索的時候負載將會集中在個別RegionServer，降低查詢效率。

Rowkey唯一原則

必須在設計上保證其唯一性。

解答：

1、在庫表設計的時候，盡量考慮rowkey和columnfamily的特性

2、進行hbase集群的調優

解答：

hbase的filter是通過scan設置的，所以是基于scan的查詢結果進行過濾。

1.在進行訂單開發的時候，我們使用rowkeyfilter過濾出某個用戶的所有訂單

2.在進行云筆記開發時，我們使用rowkey過濾器進行redis數據的恢復。

解答：

?

解答：

（1） 通過單個Rowkey訪問，即按照某個Rowkey鍵值進行get操作，這樣獲取唯一一條記錄；

（2） 通過Rowkey的range進行scan，即通過設置startRowKey和endRowKey，在這個范圍內進行掃描。這樣可以按指定的條件獲取一批記錄；

（3） 全表掃描，即直接掃描整張表中所有行記錄。

?

解答：

HBase的瓶頸就是硬盤傳輸速度。HBase的操作，它可以往數據里面insert,也可以update一些數據，但update的實際上也是insert，只是插入一個新的時間戳的一行。Delete數據，也是insert，只是insert一行帶有delete標記的一行。Hbase的所有操作都是追加插入操作。Hbase是一種日志集數據庫。它的存儲方式，像是日志文件一樣。它是批量大量的往硬盤中寫，通常都是以文件形式的讀寫。這個讀寫速度，就取決于硬盤與機器之間的傳輸有多快。而Oracle的瓶頸是硬盤尋道時間。它經常的操作時隨機讀寫。要update一個數據，先要在硬盤中找到這個block，然后把它讀入內存，在內存中的緩存中修改，過段時間再回寫回去。由于你尋找的block不同，這就存在一個隨機的讀。硬盤的尋道時間主要由轉速來決定的。而尋道時間，技術基本沒有改變，這就形成了尋道時間瓶頸。

解答：

在HMaster、RegionServer內部，創建了RpcServer實例，并與Client三者之間實現了Rpc調用，HBase0.95內部引入了Google-Protobuf作為中間數據組織方式，并在Protobuf提供的Rpc接口之上，實現了基于服務的Rpc實現，本文詳細闡述了HBase-Rpc實現細節。

HBase的RPC Protocol

?在HMaster、RegionServer內部，實現了rpc 多個protocol來完成管理和應用邏輯，具體如下protocol如下：

HMaster支持的Rpc協議：
MasterMonitorProtocol，Client與Master之間的通信，Master是RpcServer端，主要實現HBase集群監控的目的。

MasterAdminProtocol，Client與Master之間的通信，Master是RpcServer端，主要實現HBase表格的管理。例如TableSchema的更改，Table-Region的遷移、合并、下線(Offline)、上線(Online)以及負載平衡，以及Table的刪除、快照等相關功能。

RegionServerStatusProtoco，RegionServer與Master之間的通信，Master是RpcServer端，負責提供RegionServer向HMaster狀態匯報的服務。

RegionServer支持的Rpc協議：

ClientProtocol，Client與RegionServer之間的通信，RegionServer是RpcServer端，主要實現用戶的讀寫請求。例如get、multiGet、mutate、scan、bulkLoadHFile、執行Coprocessor等。

AdminProtocols，Client與RegionServer之間的通信，RegionServer是RpcServer端，主要實現Region、服務、文件的管理。例如storefile信息、Region的操作、WAL操作、Server的開關等。

(備注：以上提到的Client可以是用戶Api、也可以是RegionServer或者HMaster)

?

?HBase-RPC實現機制分析

RpcServer配置三個隊列：

1）普通隊列callQueue，絕大部分Call請求存在該隊列中：callQueue上maxQueueLength為${ipc.server.max.callqueue.length},默認是${hbase.master.handler.count}*DEFAULT_MAX_CALLQUEUE_LENGTH_PER_HANDLER，目前0.95.1中，每個Handler上CallQueue的最大個數默認值(DEFAULT_MAX_CALLQUEUE_LENGTH_PER_HANDLER)為10。

2）優先級隊列: PriorityQueue。如果設置priorityHandlerCount的個數，會創建與callQueue相當容量的queue存儲Call，該優先級隊列對應的Handler的個數由rpcServer實例化時傳入。

3）拷貝隊列：replicationQueue。由于RpcServer由HMaster和RegionServer共用，該功能僅為RegionServer提供，queue的大小為${ipc.server.max.callqueue.size}指定，默認為1024*1024*1024，handler的個數為hbase.regionserver.replication.handler.count。

RpcServer由三個模塊組成：

Listener ===Queue=== Responder

?

這里以HBaseAdmin.listTables為例，???? 分析一個Rpc請求的函數調用過程：

1) RpcClient創建一個BlockingRpcChannel。

2）以channel為參數創建執行RPC請求需要的stub，此時的stub已經被封裝在具體Service下，stub下定義了可執行的rpc接口。

3）stub調用對應的接口，實際內部channel調用callBlockingMethod方法。

通過以上的實現細節，最終轉換成rpcClient的調用，使用MethodDescriptor封裝了不同rpc函數，使用Message基類可以接收基于Message的不同的Request和Response對象。

4）RpcClient創建Call對象，查找或者創建合適的Connection，并喚醒Connection。

5）Connection等待Call的Response，同時rpcClient調用函數中，會使用connection.writeRequest(Call call)將請求寫入到RpcServer網絡流中。

6）等待Call的Response，然后層層返回給更上層接口，從而完成此次RPC調用。

RPCServer收到的Rpc報文的內部組織如下：

整個Request存儲是經過編碼之后的byte數組，包括如下幾個部分：

從功能上講，RpcServer上包含了三個模塊，

1）Listener。包含了多個Reader線程，通過Selector獲取ServerSocketChannel接收來自RpcClient發送來的Connection，并從中重構Call實例，添加到CallQueue隊列中。

?”IPC Server listener on 60021″ daemon prio=10 tid=0x00007f7210a97800 nid=0x14c6 runnable [0x00007f720e8d0000]
java.lang.Thread.State: RUNNABLE
at sun.nio.ch.EPollArrayWrapper.epollWait(Native Method)
at sun.nio.ch.EPollArrayWrapper.poll(EPollArrayWrapper.java:210)
at sun.nio.ch.EPollSelectorImpl.doSelect(EPollSelectorImpl.java:65)
at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:69)
- locked <0x00000000c43cae68> (a sun.nio.ch.Util$2)
- locked <0x00000000c43cae50> (a java.util.Collections$UnmodifiableSet)
- locked <0x00000000c4322ca8> (a sun.nio.ch.EPollSelectorImpl)
at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:80)
at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:84)
at org.apache.hadoop.hbase.ipc.RpcServer$Listener.run(RpcServer.java:646)

2）Handler。負責執行Call，調用Service的方法，然后返回Pair<Message,CellScanner>

“IPC Server handler 0 on 60021″ daemon prio=10 tid=0x00007f7210eab000 nid=0x14c7 waiting on condition [0x00007f720e7cf000]
java.lang.Thread.State: WAITING (parking)
at sun.misc.Unsafe.park(Native Method)
- parking to wait for? <0x00000000c43cad90> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
at java.util.concurrent.locks.LockSupport.park(LockSupport.java:156)
at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:1987)
at java.util.concurrent.LinkedBlockingQueue.take(LinkedBlockingQueue.java:399)
at org.apache.hadoop.hbase.ipc.RpcServer$Handler.run(RpcServer.java:1804)

3) Responder。負責把Call的結果返回給RpcClient。

?”IPC Server Responder” daemon prio=10 tid=0x00007f7210a97000 nid=0x14c5 runnable [0x00007f720e9d1000]
java.lang.Thread.State: RUNNABLE
at sun.nio.ch.EPollArrayWrapper.epollWait(Native Method)
at sun.nio.ch.EPollArrayWrapper.poll(EPollArrayWrapper.java:210)
at sun.nio.ch.EPollSelectorImpl.doSelect(EPollSelectorImpl.java:65)
at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:69)
- locked <0x00000000c4407078> (a sun.nio.ch.Util$2)
- locked <0x00000000c4407060> (a java.util.Collections$UnmodifiableSet)
- locked <0x00000000c4345b68> (a sun.nio.ch.EPollSelectorImpl)
at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:80)
at org.apache.hadoop.hbase.ipc.RpcServer$Responder.doRunLoop(RpcServer.java:833)
at org.apache.hadoop.hbase.ipc.RpcServer$Responder.run(RpcServer.java:816)

RpcClient為Rpc請求建立Connection，通過Connection將Call發送RpcServer，然后RpcClient等待結果的返回。

?

解答：

解答：

拆分復雜的業務到多個bolt中，這樣可以利用bolt的tree將速度提升

提高并行度

解答：

開發流程：

寫主類（設計spout和bolt的分發機制）

寫spout收集數據

寫bolt處理數據，根據數據量和業務的復雜程度，設計并行度。

?

容錯機制：

采用ack和fail進行容錯，失敗的數據重新發送。

解答：

?

解答：

1、Flume的channel分為很多種，可以將數據寫入到文件

2、防止非首個agent宕機的方法數可以做集群或者主備

?

解答：

Flume的配置圍繞著source、channel、sink敘述，flume的集群是做在agent上的，而非機器上。

?

解答：

優點：Nginx的日志格式是固定的，但是缺少sessionid，通過logger4j采集的日志是帶有sessionid的，而session可以通過redis共享，保證了集群日志中的同一session落到不同的tomcat時，sessionId還是一樣的，而且logger4j的方式比較穩定，不會宕機。

缺點：不夠靈活，logger4j的方式和項目結合過于緊密，而flume的方式比較靈活，拔插式比較好，不會影響項目性能。

解答：

Flume采集日志是通過流的方式直接將日志收集到存儲層，而kafka試講日志緩存在kafka集群，待后期可以采集到存儲層。

Flume采集中間停了，可以采用文件的方式記錄之前的日志，而kafka是采用offset的方式記錄之前的日志。

解答：

分區備份，存在主備partition

解答：

Producer à leader partition à follower partition(半數以上) àconsumer

解答：

spark streaming從1.2開始提供了數據的零丟失，想享受這個特性，需要滿足如下條件：

數據輸入需要可靠的sources和可靠的receivers

應用metadata必須通過應用driver checkpoint

WAL（write ahead log）

可靠的sources和receivers

spark streaming可以通過多種方式作為數據sources（包括kafka），輸入數據通過receivers接收，通過replication存儲于spark中（為了faultolerance，默認復制到兩個spark executors），如果數據復制完成，receivers可以知道（例如kafka中更新offsets到zookeeper中）。這樣當receivers在接收數據過程中crash掉，不會有數據丟失，receivers沒有復制的數據，當receiver恢復后重新接收。

metadata checkpoint

可靠的sources和receivers，可以使數據在receivers失敗后恢復，然而在driver失敗后恢復是比較復雜的，一種方法是通過checkpoint metadata到HDFS或者S3。metadata包括：

configuration

code

一些排隊等待處理但沒有完成的RDD（僅僅是metadata，而不是data）

這樣當driver失敗時，可以通過metadata checkpoint，重構應用程序并知道執行到那個地方。

數據可能丟失的場景

可靠的sources和receivers，以及metadata checkpoint也不可以保證數據的不丟失，例如：

兩個executor得到計算數據，并保存在他們的內存中

receivers知道數據已經輸入

executors開始計算數據

driver突然失敗

driver失敗，那么executors都會被kill掉

因為executor被kill掉，那么他們內存中得數據都會丟失，但是這些數據不再被處理

executor中的數據不可恢復

WAL

為了避免上面情景的出現，spark streaming 1.2引入了WAL。所有接收的數據通過receivers寫入HDFS或者S3中checkpoint目錄，這樣當driver失敗后，executor中數據丟失后，可以通過checkpoint恢復。

At-Least-Once

盡管WAL可以保證數據零丟失，但是不能保證exactly-once，例如下面場景：

Receivers接收完數據并保存到HDFS或S3

在更新offset前，receivers失敗了

Spark Streaming以為數據接收成功，但是Kafka以為數據沒有接收成功，因為offset沒有更新到zookeeper

隨后receiver恢復了

從WAL可以讀取的數據重新消費一次，因為使用的kafka High-Level消費API，從zookeeper中保存的offsets開始消費

WAL的缺點

通過上面描述，WAL有兩個缺點：

降低了receivers的性能，因為數據還要存儲到HDFS等分布式文件系統

對于一些resources，可能存在重復的數據，比如Kafka，在Kafka中存在一份數據，在Spark Streaming也存在一份（以WAL的形式存儲在hadoop API兼容的文件系統中）

Kafka direct API

為了WAL的性能損失和exactly-once，spark streaming1.3中使用Kafka direct API。非常巧妙，Spark driver計算下個batch的offsets，指導executor消費對應的topics和partitions。消費Kafka消息，就像消費文件系統文件一樣。

不再需要kafka receivers，executor直接通過Kafka API消費數據

WAL不再需要，如果從失敗恢復，可以重新消費

exactly-once得到了保證，不會再從WAL中重復讀取數據

總結

主要說的是spark streaming通過各種方式來保證數據不丟失，并保證exactly-once，每個版本都是spark streaming越來越穩定，越來越向生產環境使用發展。

?

解答：

1、topic是按照“主題名-分區”存儲的

2、分區個數由配置文件決定

3、每個分區下最重要的兩個文件是0000000000.log和000000.index，0000000.log以默認1G大小回滾。

解答：

Mr是文件方式的分布式計算框架，是將中間結果和最終結果記錄在文件中，map和reduce的數據分發也是在文件中。

spark是內存迭代式的計算框架，計算的中間結果可以緩存內存，也可以緩存硬盤，但是不是每一步計算都需要緩存的。

Spark-rdd是一個數據的分區記錄集合………………

解答：

1、spark應用提交后，經歷了一系列的轉換，最后成為task在每個節點上執行

2、RDD的Action算子觸發Job的提交，生成RDD DAG

3、由DAGScheduler將RDD DAG轉化為Stage DAG，每個Stage中產生相應的Task集合

4、TaskScheduler將任務分發到Executor執行

5、每個任務對應相應的一個數據塊，只用用戶定義的函數處理數據塊

解答：

通過org.apache.spark.deploy.Client類執行作業，作業運行命令如下：

作業執行流程描述：

1、客戶端提交作業給Master

2、Master讓一個Worker啟動Driver，即SchedulerBackend。Worker創建一個DriverRunner線程，DriverRunner啟動SchedulerBackend進程。

3、另外Master還會讓其余Worker啟動Exeuctor，即ExecutorBackend。Worker創建一個ExecutorRunner線程，ExecutorRunner會啟動ExecutorBackend進程。

4、ExecutorBackend啟動后會向Driver的SchedulerBackend注冊。SchedulerBackend進程中包含DAGScheduler，它會根據用戶程序，生成執行計劃，并調度執行。對于每個stage的task，都會被存放到TaskScheduler中，ExecutorBackend向SchedulerBackend匯報的時候把TaskScheduler中的task調度到ExecutorBackend執行。

5、所有stage都完成后作業結束。

?

解答：

作業執行流程描述：

1、客戶端啟動后直接運行用戶程序，啟動Driver相關的工作：DAGScheduler和BlockManagerMaster等。

2、客戶端的Driver向Master注冊。

3、Master還會讓Worker啟動Exeuctor。Worker創建一個ExecutorRunner線程，ExecutorRunner會啟動ExecutorBackend進程。

4、ExecutorBackend啟動后會向Driver的SchedulerBackend注冊。Driver的DAGScheduler解析作業并生成相應的Stage，每個Stage包含的Task通過TaskScheduler分配給Executor執行。

5、所有stage都完成后作業結束。

sqoop import --connect jdbc:mysql://192.168.56.204:3306/sqoop --username hive --password hive --table jobinfo --target-dir /sqoop/test7 --inline-lob-limit 16777216 --fields-terminated-by '\t' -m 2

?

sqoop create-hive-table --connect jdbc:mysql://192.168.56.204:3306/sqoop --table jobinfo --username hive --password hive --hive-table sqtest --fields-terminated-by "\t" --lines-terminated-by "\n";

解答：

Sqoop是一個用來將Hadoop和關系型數據庫中的數據相互轉移的工具，可以將一個關系型數據庫（例如 ： MySQL ,Oracle ,Postgres等）中的數據導進到Hadoop的HDFS中，也可以將HDFS的數據導進到關系型數據庫中。

?

首先需以下要準備：

第一：hadoop的NameNode節點下lib文件夾中要有相應數據庫驅動的jar包和sqoop的jar包。

第二：預先在相應的數據庫創建Table，注：在HDFS的某個目錄上的數據格式要和相應的表中的字段數量一致。

?

由于我這里使用的是Oracle數據庫并且是使用Java來操作的。所以下面的代碼以及截圖都是以Java的例子：

首先標準化HDFS中文件格式，如下圖：

?

Java代碼如下：

Configuration conf = new Configuration();
conf.set("fs.default.name", "hdfs://192.168.115.5:9000");
conf.set("hadoop.job.ugi", "hadooper,hadoopgroup");
conf.set("mapred.job.tracker", "192.168.115.5:9001");


ArrayList<String> list = new ArrayList<String>(); // 定義一個list
list.add("--table");
list.add("A_BAAT_CLIENT"); // Oracle中的表。將來數據要導入到這個表中。
list.add("--export-dir");
list.add("/home/hadoop/traffic/capuse/near7date/activeUser/capuse_near7_activeUser_2013-02-06.log"); // hdfs上的目錄。這個目錄下的數據要導入到a_baat_client這個表中。
list.add("--connect");
list.add("jdbc:oracle:thin:@10.18.96.107:1521:life"); // Oracle的鏈接
list.add("--username");
list.add("TRAFFIC"); // Oracle的用戶名
list.add("--password");
list.add("TRAFFIC"); // Oracle的密碼
list.add("--input-fields-terminated-by");
list.add("|"); // 數據分隔符號
list.add("-m");
list.add("1");// 定義mapreduce的數量。


String[] arg = new String[1];
ExportTool exporter = new ExportTool();
Sqoop sqoop = new Sqoop(exporter);
sqoop.setConf(conf);
arg = list.toArray(new String[0]);
int result = Sqoop.runSqoop(sqoop, arg);
System.out.println("res:" + result); // 打印執行結果。

?

最后再在Main方法中運行即可，生成后表數據如下圖所示：

通過上面的操作以及代碼即可在Java中實現把HDFS數據生成對應的表數據；

不過除了可以用Java來實現，使用基本的命令也是可以的，命令如下：

在Hadoop bin目錄中：

sqoop export --connect jdbc:oracle:thin:@10.18.96.107:1521:life \

--table A_BAAT_CLIENT --username TRAFFIC --password TRAFFIC \
--input-fields-terminated-by '|' \
--export-dir /home/hadoop/traffic/capuse/near7date/activeUser/test.log ?-m 1

意思和上面Java中代碼一樣。

?

注意：

1、數據庫表名、用戶名、密碼使用大寫（這有可能會出現問題，因為我在測試過程中，使用小寫時出現錯誤，出現No Columns這個經典錯誤。所以推薦大寫，當然這不是必須）；

2、預先建好相應的Table；

?

解答：

redis：分布式緩存，強調緩存，內存中數據

傳統數據庫：注重關系

hbase：列式數據庫，無法做關系數據庫的主外鍵，用于存儲海量數據，底層基于hdfs

hive：數據倉庫工具，底層是mapreduce。不是數據庫，不能用來做用戶的交互存儲

?

解答：

倒排索引(Inverted index)

適用范圍：搜索引擎，關鍵字查詢

基本原理及要點：為何叫倒排索引？一種索引方法，被用來存儲在全文搜索下某個單詞在一個文檔或者一組文檔中的存儲位置的映射。

以英文為例，下面是要被索引的文本：

T0 = “it is what it is”

T1 = “what is it”

T2 = “it is a banana”

我們就能得到下面的反向文件索引：

“a”: {2}

“banana”: {2}

“is”: {0, 1, 2}

“it”: {0, 1, 2}

“what”: {0, 1}

檢索的條件”what”,”is”和”it”將對應集合的交集。

正向索引開發出來用來存儲每個文檔的單詞的列表。正向索引的查詢往往滿足每個文檔有序 頻繁的全文查詢和每個單詞在校驗文檔中的驗證這樣的查詢。在正向索引中，文檔占據了中心的位置，每個文檔指向了一個它所包含的索引項的序列。也就是說文檔指向了它包含的那些單詞，而反向索引則是單詞指向了包含它的文檔，很容易看到這個反向的關系。

?

解答：

數據庫范式1NF 2NF 3NF BCNF(實例）
??? 設計范式（范式,數據庫設計范式,數據庫的設計范式）是符合某一種級別的關系模式的集合。構造數據庫必須遵循一定的規則。在關系數據庫中，這種規則就是范式。關系數據庫中的關系必須滿足一定的要求，即滿足不同的范式。目前關系數據庫有六種范式：第一范式（1NF）、第二范式（2NF）、第三范式（3NF）、第四范式（4NF）、第五范式（5NF）和第六范式（6NF）。滿足最低要求的范式是第一范式（1NF）。在第一范式的基礎上進一步滿足更多要求的稱為第二范式（2NF），其余范式以次類推。一般說來，數據庫只需滿足第三范式（3NF）就行了。下面我們舉例介紹第一范式（1NF）、第二范式（2NF）和第三范式（3NF）。
??? 在創建一個數據庫的過程中，范化是將其轉化為一些表的過程，這種方法可以使從數據庫得到的結果更加明確。這樣可能使數據庫產生重復數據，從而導致創建多余的表。范化是在識別數據庫中的數據元素、關系，以及定義所需的表和各表中的項目這些初始工作之后的一個細化的過程。
??? 下面是范化的一個例子 Customer Item purchased Purchase price Thomas Shirt $40 Maria Tennis shoes $35 Evelyn Shirt $40 Pajaro Trousers $25
如果上面這個表用于保存物品的價格，而你想要刪除其中的一個顧客，這時你就必須同時刪除一個價格。范化就是要解決這個問題，你可以將這個表化為兩個表，一個用于存儲每個顧客和他所買物品的信息，另一個用于存儲每件產品和其價格的信息，這樣對其中一個表做添加或刪除操作就不會影響另一個表。

關系數據庫的幾種設計范式介紹

1 第一范式（1NF）
??? 在任何一個關系數據庫中，第一范式（1NF）是對關系模式的基本要求，不滿足第一范式（1NF）的數據庫就不是關系數據庫。
??? 所謂第一范式（1NF）是指數據庫表的每一列都是不可分割的基本數據項，同一列中不能有多個值，即實體中的某個屬性不能有多個值或者不能有重復的屬性。如果出現重復的屬性，就可能需要定義一個新的實體，新的實體由重復的屬性構成，新實體與原實體之間為一對多關系。在第一范式（1NF）中表的每一行只包含一個實例的信息。例如，對于圖3-2 中的員工信息表，不能將員工信息都放在一列中顯示，也不能將其中的兩列或多列在一列中顯示；員工信息表的每一行只表示一個員工的信息，一個員工的信息在表中只出現一次。簡而言之，第一范式就是無重復的列。

2 第二范式（2NF）
??? 第二范式（2NF）是在第一范式（1NF）的基礎上建立起來的，即滿足第二范式（2NF）必須先滿足第一范式（1NF）。第二范式（2NF）要求數據庫表中的每個實例或行必須可以被惟一地區分。為實現區分通常需要為表加上一個列，以存儲各個實例的惟一標識。如圖3-2 員工信息表中加上了員工編號（emp_id）列，因為每個員工的員工編號是惟一的，因此每個員工可以被惟一區分。這個惟一屬性列被稱為主關鍵字或主鍵、主碼。
第二范式（2NF）要求實體的屬性完全依賴于主關鍵字。所謂完全依賴是指不能存在僅依賴主關鍵字一部分的屬性，如果存在，那么這個屬性和主關鍵字的這一部分應該分離出來形成一個新的實體，新實體與原實體之間是一對多的關系。為實現區分通常需要為表加上一個列，以存儲各個實例的惟一標識。簡而言之，第二范式就是非主屬性非部分依賴于主關鍵字。

3 第三范式（3NF）
??? 滿足第三范式（3NF）必須先滿足第二范式（2NF）。簡而言之，第三范式（3NF）要求一個數據庫表中不包含已在其它表中已包含的非主關鍵字信息。例如，存在一個部門信息表，其中每個部門有部門編號（dept_id）、部門名稱、部門簡介等信息。那么在圖3-2的員工信息表中列出部門編號后就不能再將部門名稱、部門簡介等與部門有關的信息再加入員工信息表中。如果不存在部門信息表，則根據第三范式（3NF）也應該構建它，否則就會有大量的數據冗余。簡而言之，第三范式就是屬性不依賴于其它非主屬性。
數據庫設計三大范式應用實例剖析
???? 數據庫的設計范式是數據庫設計所需要滿足的規范，滿足這些規范的數據庫是簡潔的、結構明晰的，同時，不會發生插入（insert）、刪除（delete）和更新（update）操作異常。反之則是亂七八糟，不僅給數據庫的編程人員制造麻煩，而且面目可憎，可能存儲了大量不需要的冗余信息。
??? 設計范式是不是很難懂呢？非也，大學教材上給我們一堆數學公式我們當然看不懂，也記不住。所以我們很多人就根本不按照范式來設計數據庫。
實質上，設計范式用很形象、很簡潔的話語就能說清楚，道明白。本文將對范式進行通俗地說明，并以筆者曾經設計的一個簡單論壇的數據庫為例來講解怎樣將這些范式應用于實際工程。

范式說明
??? 第一范式（1NF）：數據庫表中的字段都是單一屬性的，不可再分。這個單一屬性由基本類型構成，包括整型、實數、字符型、邏輯型、日期型等。

??? 例如，如下的數據庫表是符合第一范式的：

??? 字段1 字段2 字段3 字段4

??? 而這樣的數據庫表是不符合第一范式的：

??? 字段1 字段2 字段3 字段4
??? 字段3.1 字段3.2?

??? 很顯然，在當前的任何關系數據庫管理系統（DBMS）中，傻瓜也不可能做出不符合第一范式的數據庫，因為這些DBMS不允許你把數據庫表的一列再分成二列或多列。因此，你想在現有的DBMS中設計出不符合第一范式的數據庫都是不可能的。?
??? 第二范式（2NF）：數據庫表中不存在非關鍵字段對任一候選關鍵字段的部分函數依賴（部分函數依賴指的是存在組合關鍵字中的某些字段決定非關鍵字段的情況），也即所有非關鍵字段都完全依賴于任意一組候選關鍵字。

??? 假定選課關系表為SelectCourse(學號, 姓名, 年齡, 課程名稱, 成績, 學分)，關鍵字為組合關鍵字(學號, 課程名稱)，因為存在如下決定關系：
??? (學號, 課程名稱) → (姓名, 年齡, 成績, 學分)?

??? 這個數據庫表不滿足第二范式，因為存在如下決定關系：
??? (課程名稱) → (學分)
??? (學號) → (姓名, 年齡)
??? 即存在組合關鍵字中的字段決定非關鍵字的情況。?
??? 由于不符合2NF，這個選課關系表會存在如下問題：
??? (1) 數據冗余：
??? 同一門課程由n個學生選修，"學分"就重復n-1次；同一個學生選修了m門課程，姓名和年齡就重復了m-1次。
??? (2) 更新異常：
??? 若調整了某門課程的學分，數據表中所有行的"學分"值都要更新，否則會出現同一門課程學分不同的情況。
??? (3) 插入異常：
??? 假設要開設一門新的課程，暫時還沒有人選修。這樣，由于還沒有"學號"關鍵字，課程名稱和學分也無法記錄入數據庫。
??? (4) 刪除異常：
??? 假設一批學生已經完成課程的選修，這些選修記錄就應該從數據庫表中刪除。但是，與此同時，課程名稱和學分信息也被刪除了。很顯然，這也會導致插入異常。
??? 把選課關系表SelectCourse改為如下三個表：
??? 學生：Student(學號, 姓名, 年齡)；
??? 課程：Course(課程名稱, 學分)；
??? 選課關系：SelectCourse(學號, 課程名稱, 成績)。
??? 這樣的數據庫表是符合第二范式的， 消除了數據冗余、更新異常、插入異常和刪除異常。
??? 另外，所有單關鍵字的數據庫表都符合第二范式，因為不可能存在組合關鍵字。

??? 第三范式（3NF）：在第二范式的基礎上，數據表中如果不存在非關鍵字段對任一候選關鍵字段的傳遞函數依賴則符合第三范式。所謂傳遞函數依賴，指的是如果存在"A → B → C"的決定關系，則C傳遞函數依賴于A。因此，滿足第三范式的數據庫表應該不存在如下依賴關系：
??? 關鍵字段 → 非關鍵字段x → 非關鍵字段y
??? 假定學生關系表為Student(學號, 姓名, 年齡, 所在學院, 學院地點, 學院電話)，關鍵字為單一關鍵字"學號"，因為存在如下決定關系：
??? (學號) → (姓名, 年齡, 所在學院, 學院地點, 學院電話)
??? 這個數據庫是符合2NF的，但是不符合3NF，因為存在如下決定關系：
??? (學號) → (所在學院) → (學院地點, 學院電話)
??? 即存在非關鍵字段"學院地點"、"學院電話"對關鍵字段"學號"的傳遞函數依賴。 ??
??? 它也會存在數據冗余、更新異常、插入異常和刪除異常的情況，讀者可自行分析得知。
??? 把學生關系表分為如下兩個表：
??? 學生：(學號, 姓名, 年齡, 所在學院)；
?? 學院：(學院, 地點, 電話)。?

??? 這樣的數據庫表是符合第三范式的，消除了數據冗余、更新異常、插入異常和刪除異常。
鮑依斯-科得范式（BCNF）：在第三范式的基礎上，數據庫表中如果不存在任何字段對任一候選關鍵字段的傳遞函數依賴則符合第三范式。
??? 假設倉庫管理關系表為StorehouseManage(倉庫ID, 存儲物品ID, 管理員ID, 數量)，且有一個管理員只在一個倉庫工作；一個倉庫可以存儲多種物品。這個數據庫表中存在如下決定關系：
??? (倉庫ID, 存儲物品ID) →(管理員ID, 數量)
??? (管理員ID, 存儲物品ID) → (倉庫ID, 數量)
??? 所以，(倉庫ID, 存儲物品ID)和(管理員ID, 存儲物品ID)都是StorehouseManage的候選關鍵字，表中的唯一非關鍵字段為數量，它是符合第三范式的。但是，由于存在如下決定關系：
??? (倉庫ID) → (管理員ID)
??? (管理員ID) → (倉庫ID)
??? 即存在關鍵字段決定關鍵字段的情況，所以其不符合BCNF范式。它會出現如下異常情況：
??? (1) 刪除異常：
??? 當倉庫被清空后，所有"存儲物品ID"和"數量"信息被刪除的同時，"倉庫ID"和"管理員ID"信息也被刪除了。
??? (2) 插入異常：
??? 當倉庫沒有存儲任何物品時，無法給倉庫分配管理員。
??? (3) 更新異常：
??? 如果倉庫換了管理員，則表中所有行的管理員ID都要修改。
??? 把倉庫管理關系表分解為二個關系表：
??? 倉庫管理：StorehouseManage(倉庫ID, 管理員ID)；
??? 倉庫：Storehouse(倉庫ID, 存儲物品ID, 數量)。
??? 這樣的數據庫表是符合BCNF范式的，消除了刪除異常、插入異常和更新異常。
范式應用
??? 我們來逐步搞定一個論壇的數據庫，有如下信息：
??? （1） 用戶：用戶名，email，主頁，電話，聯系地址
??? （2） 帖子：發帖標題，發帖內容，回復標題，回復內容
??? 第一次我們將數據庫設計為僅僅存在表：
??? 用戶名 email 主頁 電話 聯系地址 發帖標題 發帖內容 回復標題 回復內容
??? 這個數據庫表符合第一范式，但是沒有任何一組候選關鍵字能決定數據庫表的整行，唯一的關鍵字段用戶名也不能完全決定整個元組。我們需要增加"發帖ID"、"回復ID"字段，即將表修改為：
??? 用戶名 email 主頁 電話 聯系地址 發帖ID 發帖標題 發帖內容 回復ID 回復標題 回復內容?
??? 這樣數據表中的關鍵字(用戶名，發帖ID，回復ID)能決定整行：
??? (用戶名,發帖ID,回復ID) → (email,主頁,電話,聯系地址,發帖標題,發帖內容,回復標題,回復內容)
??? 但是，這樣的設計不符合第二范式，因為存在如下決定關系：
??? (用戶名) → (email,主頁,電話,聯系地址)
??? (發帖ID) → (發帖標題,發帖內容)
??? (回復ID) → (回復標題,回復內容)
??? 即非關鍵字段部分函數依賴于候選關鍵字段，很明顯，這個設計會導致大量的數據冗余和操作異常。
我們將數據庫表分解為（帶下劃線的為關鍵字）：
（1） 用戶信息：用戶名，email，主頁，電話，聯系地址
（2） 帖子信息：發帖ID，標題，內容
（3） 回復信息：回復ID，標題，內容
（4） 發貼：用戶名，發帖ID
（5） 回復：發帖ID，回復ID
??? 這樣的設計是滿足第1、2、3范式和BCNF范式要求的，但是這樣的設計是不是最好的呢？
不一定。
??? 觀察可知，第4項"發帖"中的"用戶名"和"發帖ID"之間是1：N的關系，因此我們可以把"發帖"合并到第2項的"帖子信息"中；第5項"回復"中的"發帖ID"和"回復ID"之間也是1：N的關系，因此我們可以把"回復"合并到第3項的"回復信息"中。這樣可以一定量地減少數據冗余，新的設計為：
（1） 用戶信息：用戶名，email，主頁，電話，聯系地址
（2） 帖子信息：用戶名，發帖ID，標題，內容
（3） 回復信息：發帖ID，回復ID，標題，內容
??? 數據庫表1顯然滿足所有范式的要求；
??? 數據庫表2中存在非關鍵字“標題”、“內容”對關鍵字段“發帖ID”的部分函數依賴，即不滿足第二范式的要求，但是這一設計并不會導致數據冗余和操作異常；
??? 數據庫表3中也存在非關鍵字段"標題"、"內容"對關鍵字段"回復ID"的部分函數依賴，也不滿足第二范式的要求，但是與數據庫表2相似，這一設計也不會導致數據冗余和操作異常。
??? 由此可以看出，并不一定要強行滿足范式的要求，對于1：N關系，當1的一邊合并到N的那邊后，N的那邊就不再滿足第二范式了，但是這種設計反而比較好！
??? 對于M：N的關系，不能將M一邊或N一邊合并到另一邊去，這樣會導致不符合范式要求，同時導致操作異常和數據冗余。
??? 對于1：1的關系，我們可以將左邊的1或者右邊的1合并到另一邊去，設計導致不符合范式要求，但是并不會導致操作異常和數據冗余。

?

解答：

方案 1：將大文件分成能夠被內存加載的小文件。

可以估計每個文件安的大小為 50G×64=320G，遠遠大于內存限制的 4G。所以不可能將其完全加載到內存中處理。考慮采取分而治之的方法。

s 遍歷文件 a，對每個 url 求取 ，然后根據所取得的值將 url 分別存儲到 1000 個小文件（記為 ）中。

這樣每個小文件的大約為 300M。

s 遍歷文件 b，采取和 a 相同的方式將 url 分別存儲到 1000 各小文件（記為 ）。這樣處理后，所有可能相同的 url 都在對應的小文件（ ）中，不對應的小文件不可能有相同的 url。然后我們只要求出 1000 對小文件中相同的 url 即可。

s 求每對小文件中相同的 url 時，可以把其中一個小文件的 url 存儲到 hash_set 中。然后遍歷另一個小文件的每個 url，看其是否在剛才構建的 hash_set 中，如果是，那么就是共同的 url，存到文件里面就可以了。

方案 2：內存映射成 BIT 最小存儲單元。

如果允許有一定的錯誤率，可以使用 Bloom filter，4G 內存大概可以表示 340 億 bit。將其中一個文件中的url使用Bloom filter映射為這340億bit，然后挨個讀取另外一個文件的url，檢查是否與Bloom filter，如果是，那么該 url 應該是共同的 url（注意會有一定的錯誤率）。

解答：

方案 1：

s 順序讀取 10 個文件，按照 hash(query)%10 的結果將 query 寫入到另外 10 個文件（記為 ）中。這樣新生成的文件每個的大小大約也 1G（假設 hash 函數是隨機的）。

s 找一臺內存在 2G 左右的機器，依次對 用 hash_map(query, query_count)來統計每個 query 出現的次數。利用快速/堆/歸并排序按照出現次數進行排序。將排序好的 query 和對應的 query_cout 輸出到文件中。這樣得到了 10 個排好序的文件（記為 ）。

s 對 這 10 個文件進行歸并排序（內排序與外排序相結合）。

方案 2：

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

次數做快速/堆/歸并排序就可以了。

方案 3：

與方案 1 類似，但在做完 hash，分成多個文件后，可以交給多個文件來處理，采用分布式的架構來處

理（比如 MapReduce），最后再進行合并。

//一般在大文件中找出出現頻率高的，先把大文件映射成小文件，模 1000，在小文件中找到高頻的

解答：

方案 1：順序讀文件中，對于每個詞 x，取 ，然后按照該值存到 5000 個小文件（記為 ）中。這樣每個文件大概是 200k 左右。如果其中的有的文件超過了 1M 大小，還可以按照類似的方法繼續往下分，知道分解得到的小文件的大小都不超過 1M。 對每個小文件，統計每個文件中出現的詞以及相應的頻率（可以采用 trie 樹/hash_map 等），并取出出現頻率最大的 100 個詞（可以用含 100 個結 點的最小堆），并把 100詞及相應的頻率存入文件，這樣又得到了 5000 個文件。下一步就是把這 5000 個文件進行歸并（類似與歸并排序）的過程了。

方案2:

1.?????? 將文件逐行讀寫到另一個文件中，并將每行單詞全變成小寫

2.?????? 十六次循環執行，將每行單詞按照a-z寫到不同文件里

3.?????? 最后相同的單詞都寫在了通一個文件里

4.?????? 再將文件讀寫到各自另一個文件里，內容是“單詞 個數”

5.?????? 定義一個treemap，大小是100，依次插入大的，移除小的

6.?????? 最后得出結果

解答：

?

1.?????? 先根據日期在日志文件中提取出ip，根據ip哈希進行分寫N個文件。

2.?????? 采用mapreduce的word cont

方案 1：首先是這一天，并且是訪問百度的日志中的 IP 取出來，逐個寫入到一個大文件中。注意到 IP是 32 位的，最多有 個 IP。同樣可以采用映射的方法，比如模 1000，把整個大文件映射為 1000 個小文件，再找出每個小文中出現頻率最大的 IP（可以采用 hash_map 進行頻率統計，然后再找出頻率最大的幾個）及相應的頻率。然后再在這 1000 個最大的 IP 中，找出那個頻率最大的 IP，即為所求。

解答：

方案 1：采用 2-Bitmap（每個數分配 2bit，00 表示不存在，01 表示出現一次，10 表示多次，11 無意義）進 行，共需內存 內存，還可以接受。然后掃描這 2.5 億個整數，查看 Bitmap 中相對應位，如果是00 變 01，01 變 10，10 保持不變。所描完事后，查看 bitmap，把對應位是 01 的整數輸出即可。

方案 2：也可采用上題類似的方法，進行劃分小文件的方法。然后在小文件中找出不重復的整數，并排序。然后再進行歸并，注意去除重復的元素。

方案3：

1.?????? 將2.5億個整數重寫到一個文件里，內個整數占一行。

2.?????? 進行對半排序重寫到新的文件里，這樣最后2.5億個整數在文件里便是有序的了

3.?????? 讀取文本，將不重復的寫到一個新的文件里即可。

解答：

方案 1：（方法不正確，取出來得不一定是top10）

s 在每臺電腦上求出 TOP10，可以采用包含 10 個元素的堆完成（TOP10 小，用最大堆，TOP10 大，用最小堆）。比如求 TOP10 大，我們首先取前 10 個元素調整成最小堆，如果發現，然后掃描后面的數據，并與堆頂元素比較，如果比堆頂元素大，那么用該元素替換堆頂，然后再調整為最小堆。最后堆中的元 素就是 TOP10 大。

s 求出每臺電腦上的 TOP10 后，然后把這 100 臺電腦上的 TOP10 組合起來，共 1000 個數據，再利用上面類似的方法求出 TOP10 就可以了。

解答：

方案 1：（同上，方法錯誤）

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

?

正確的方法，先排除

解答：

解答：

解答：

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

解答：

解答：

解答：

1、通過flume將不同系統的日志收集到kafka中

2、通過storm實時的處理PV、UV、IP

3、通過kafka的consumer將日志生產到hbase中。

4、通過離線的mapreduce或者hive，處理hbase中的數據

解答：

1.建表

2.分組（group by）統計wordcount

select word,count(1) from table1 group by word;

?

解答：

flume：日志收集系統，主要用于系統日志的收集

kafka：消息隊列，進行消息的緩存和系統的解耦

storm：實時計算框架，進行流式的計算。

?

解答：

寫個mapreduce鏈? 用依賴關系，一共三個mapreduce，第一個處理第一個文件，第二個處理第二個文件，第三個處理前兩個的輸出結果，第一個mapreduce將文件去重，第二個mapreduce也將文件去重，第三個做wordcount，wordcount為1的結果就是不同的

?

解答：

思路：例如A，他的朋友是B\C\D\E\F\，那么BC的共同朋友就是A。所以將BC作為key，將A作為value，在map端輸出即可！其他的朋友循環處理。

import java.io.IOException;

import java.util.Set;

import java.util.StringTokenizer;

import java.util.TreeSet;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Job;

10. import org.apache.hadoop.mapreduce.Mapper;

11. import org.apache.hadoop.mapreduce.Reducer;

12. import org.apache.hadoop.mapreduce.Mapper.Context;

13. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

14. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

15. import org.apache.hadoop.util.GenericOptionsParser;

?

public class FindFriend {

?

????????? public static class ChangeMapper extends Mapper<Object, Text, Text,

Text>{??????????????????????

??????????????????? @Override

??????????????????? public void map(Object key, Text value, Context context) throws

IOException, InterruptedException {

????????????????????????????? StringTokenizer itr = new StringTokenizer(value.toString());

????????????????????????????????? Text owner = new Text();

?????????? ???????????????????????Set<String> set = new TreeSet<String>();

????????????????????????????? owner.set(itr.nextToken());

????????????????????????????? while (itr.hasMoreTokens()) {

????????????????????????????????????? set.add(itr.nextToken());

????????????????????????????? }?????????????

???????????????????????????? String[] friends = new String[set.size()];

???????????????????????????? friends = set.toArray(friends);

??????????????????????????????

???????????????????????????? for(int i=0;i<friends.length;i++){

????????????????????????????????????? for(int j=i+1;j<friends.length;j++){

????????????????????????????????????????????? String outputkey = friends[i]+friends[j];

????????????????????????????????????????????? context.write(new Text(outputkey),owner);

????????????????????????????????????? }?????????????????????????????????????

????????????????????????????? }

??????????????????? }

?????????? }

???????????

?????????? public static class FindReducer extends Reducer<Text,Text,Text,Text>

{??????????????????????????

???????????????????????? public void reduce(Text key, Iterable<Text> values,?

???????????????????????????????????????? Context context) throws IOException,

InterruptedException {

?????????????????????????????????? String? commonfriends ="";?

?????????????????????????????? for (Text val : values) {

????????????????????????????????? if(commonfriends == ""){

????????????????????????????????????????? commonfriends = val.toString();

?????????????????????????????????? }else{

????????????????????????????????????????? commonfriends =

commonfriends+":"+val.toString();

?????????????????????????????????? }

??????????????????????????????? }

????????????????????????????? context.write(key, new

Text(commonfriends));????????????????????????????????

???????????????????????? }??????????????????????????

?????????? }

??????????

?

??????? public static void main(String[] args) throws IOException,

???????? InterruptedException, ClassNotFoundException {

?????????????????

??????????? Configuration conf = new Configuration();

???????????? String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();

???????????? if (otherArgs.length < 2) {

?????????????? System.err.println("args error");

?????????????? System.exit(2);

??????????? }

???????????? Job job = new Job(conf, "word count");

??????????? job.setJarByClass(FindFriend.class);

???????????? job.setMapperClass(ChangeMapper.class);

???????????? job.setCombinerClass(FindReducer.class);

???????????? job.setReducerClass(FindReducer.class);

???????????? job.setOutputKeyClass(Text.class);

???????????? job.setOutputValueClass(Text.class);

???????????? for (int i = 0; i < otherArgs.length - 1; ++i) {

?????????????? FileInputFormat.addInputPath(job, new Path(otherArgs[i]));

???????????? }

???????????? FileOutputFormat.setOutputPath(job,

?????????????? new Path(otherArgs[otherArgs.length - 1]));

???????????? System.exit(job.waitForCompletion(true) ? 0 : 1);

?????????????????

??????? }

?

}

?

結果：

1. AB????? E:C:D

2. AC????? E:B

3. AD????? B:E

4. AE????? C:B:D

5. BC????? A:E

6. BD????? A:E

7. BE????? C:D:A

8. BF????? A

9. CD????? E:A:B

10. CE????? A:B

11. CF????? A

12. DE????? B:A

13. DF????? A

14. EF????? A

解答：

1.將小文件打成har文件存儲

2.將小文件序列化到hdfs中

?

解答：

ArrayList 和Vector是采用數組方式存儲數據的,是根據索引來訪問元素的，都可以根據需要自動擴展內部數據長度，以便增加和插入元素，都允許直接序號索引元素，但是插入數據要涉及到數組元素移動等內存操作，所以索引數據快插入數據慢，他們最大的區別就是synchronized同步的使用。

LinkedList使用雙向鏈表實現存儲，按序號索引數據需要進行向前或向后遍歷，但是插入數據時只需要記錄本項的前后項即可，所以插入數度較快！

如果只是查找特定位置的元素或只在集合的末端增加、移除元素，那么使用Vector或ArrayList都可以。如果是對其它指定位置的插入、刪除操作，最好選擇LinkedList

HashMap、HashTable的區別及其優缺點：

HashTable 中的方法是同步的 HashMap的方法在缺省情況下是非同步的 因此在多線程環境下需要做額外的同步機制。

HashTable不允許有null值 key和value都不允許，而HashMap允許有null值 key和value都允許 因此HashMap 使用containKey（）來判斷是否存在某個鍵。

HashTable 使用Enumeration ，而HashMap使用iterator。

Hashtable是Dictionary的子類，HashMap是Map接口的一個實現類。