源碼走讀到這個(gè)階段，其實(shí)整個(gè) Hdfs 的交互流程都已經(jīng)完全走讀完了，雖然肯定有一些細(xì)節(jié)的地方?jīng)]有進(jìn)行走讀，但是不會(huì)影響對(duì)整個(gè) Hdfs 的架構(gòu)認(rèn)知。
這一篇文章也將是整個(gè) Hdfs 源碼走讀的最后一篇，在本文中，將介紹 NameNode 、 DataNode 之外的其他節(jié)點(diǎn)信息和部分高階功能。
Federation 與 ViewFS
為了保證 Hdfs 文件系統(tǒng)中的數(shù)據(jù)一致性，對(duì)于同一個(gè)文件系統(tǒng)只會(huì)有一個(gè) NameNode 負(fù)責(zé)接收操作請(qǐng)求，即便在 HA 模式下，整個(gè)集群中也只會(huì)有一個(gè) ActiveNameNode 負(fù)責(zé)處理消息。
也正是這個(gè)原因，在 Hdfs 中存在一個(gè)節(jié)點(diǎn)數(shù)量瓶頸，類似 Android 中的 65536 的方法數(shù)限制，在這里也有一個(gè) 2^30 個(gè)理論上限(LightWeightGSet::computeCapacity中可以看到這個(gè)值的來(lái)源依據(jù))。
為了解決這個(gè)問(wèn)題，在 Hadoop 0.23.0 中引入了 Federation 的概念，用以解決 Hdfs 橫向擴(kuò)容的需求。
如上圖所示，左側(cè)是進(jìn)行聯(lián)合部署前的集群情況，NameNode 和 DataNode 之間是一對(duì)多的關(guān)系，整個(gè) Hdfs 集群中只有一個(gè) NameNode 節(jié)點(diǎn)，所有的 DataNode 節(jié)點(diǎn)都只同唯一的 NameNode 進(jìn)行通信。右側(cè)是進(jìn)行聯(lián)合部署后的集群情況，NameNode 和 DataNode 之間是多對(duì)多的關(guān)系，Hdfs 集群中同時(shí)存在多個(gè) NameNode 節(jié)點(diǎn)，每個(gè)節(jié)點(diǎn)都是 ActiveNameNode，單個(gè) DataNode 可以存放任意個(gè)數(shù)的 NameNode 的 Block 數(shù)據(jù)。
這意味著對(duì)于同一個(gè)集群，我們可以通過(guò)讓 DataNode 連接多個(gè) NameNode 使得 NameNode 的文件節(jié)點(diǎn)數(shù)量不再受限，可以進(jìn)行任意的水平擴(kuò)容。
我們知道 NameNode 中會(huì)唯一的存在一個(gè) FSDirectory 類負(fù)責(zé)記錄當(dāng)前的節(jié)點(diǎn)信息，每個(gè)文件節(jié)點(diǎn)都由一個(gè)或多個(gè) BlockInfo 構(gòu)成。為了實(shí)現(xiàn) Federation 能夠區(qū)分來(lái)自不同 NameNode 的 Block 信息，我們認(rèn)為屬于同一個(gè) NameNode 的 BlockInfo 應(yīng)該屬于同一個(gè) BlockPool, 在構(gòu)造 NameNode 的時(shí)候創(chuàng)建一個(gè)唯一的 BPId (BlockPoolID)，每個(gè) NameNode 通過(guò)不同的 BPId 進(jìn)行區(qū)分。
如上圖所示，在同一個(gè) DataNode 中，不同的 NameNode 對(duì)應(yīng)不同的 BlockPool。從實(shí)現(xiàn)邏輯上來(lái)看，在 ${DATA_DIR}/current 下會(huì)根據(jù)不同 NameNode 返回的不同 BlcokPoolId 生成對(duì)應(yīng)的文件夾，從而使得不同 NameNode 的文件數(shù)據(jù)彼此物理隔離不相互影響。
引入 Federation 之后，Hdfs 集群中存在多個(gè) NameNode 節(jié)點(diǎn)，用戶在執(zhí)行 hadoop 指令的時(shí)候必須寫(xiě)上完整的 url 路徑才能夠訪問(wèn)指定的 NameNode 節(jié)點(diǎn)，隨著集群節(jié)點(diǎn)越來(lái)越來(lái)，復(fù)雜冗余的 hdfs 地址極有可能產(chǎn)生誤操作，影響線上數(shù)據(jù)。為了解決這個(gè)問(wèn)題，Hdfs 中又引入了 ViewFS 的文件掛載概念。
在 core-site.xml 中將 fs.defaultFS 的 schema 配置為 viewfs, 并在 mountTable.xml 中配置對(duì)應(yīng)的掛載信息，即可將不同的文件系統(tǒng)掛載到 ViewFs 這個(gè)偽文件系統(tǒng)中。
如上圖所示，我們有 /User、/Work、/Tmp 三個(gè)工作目錄，分別代指不同的 NameNode 和本地文件系統(tǒng)。通過(guò)將 NameNodeA 掛載在 /User 路徑下，將 NameNodeB 掛載在 /Work 路徑下，將本地文件系統(tǒng)掛載在 /Tmp 中，我們可以使用同一個(gè) host 的不同路徑訪問(wèn)不同的文件系統(tǒng)。
@Override
protected AbstractFileSystem getTargetFileSystem(final URI uri)
throws URISyntaxException, UnsupportedFileSystemException {
String pathString = uri.getPath();
if (pathString.isEmpty()) {
pathString = "/";
}
return new ChRootedFs(
AbstractFileSystem.createFileSystem(uri, config),
new Path(pathString));
}
具體的實(shí)現(xiàn)邏輯可以查看 ViewFs 類，在 ViewFs 中處理請(qǐng)求消息時(shí)，會(huì)針對(duì)不同的掛載點(diǎn)，獲取其真實(shí)路徑，在 getTargetFileSystem 中會(huì)根據(jù)真實(shí)的 uri 地址返回對(duì)應(yīng)的 FileSystem，執(zhí)行操作。ViewFs 可以理解為一個(gè)進(jìn)行代理分發(fā)的接口，將查詢不同掛載點(diǎn)的請(qǐng)求數(shù)據(jù)，分發(fā)給對(duì)應(yīng)的 FileSystem 進(jìn)行請(qǐng)求。
SecondaryNameNode
NameNode 中為了保證文件系統(tǒng)的讀取效率，會(huì)將整個(gè)文件節(jié)點(diǎn)信息全部加載到內(nèi)存中成為 FSDirectory，所有對(duì)文件系統(tǒng)的操作，將直接在內(nèi)存中進(jìn)行更正，并通過(guò) EditLog 進(jìn)行逐一記錄到磁盤(pán)中。
如果 NameNode 掛掉重啟時(shí)，為了保證已有文件節(jié)點(diǎn)信息的完整性，會(huì)逐一對(duì) EditLog 進(jìn)行重演，從而恢復(fù)到掛掉前的狀態(tài)。
對(duì) EditLog 的重演是一個(gè)串行的過(guò)程，當(dāng)需要重演的 EditLog 數(shù)量過(guò)多時(shí)，會(huì)嚴(yán)重滯后 NameNode 的啟動(dòng)時(shí)間。為了降低重演 EditLog 的時(shí)間消耗，在 Hdfs 中產(chǎn)生了 CheckPoint 的概念。
FSDirectory 可以認(rèn)為是消費(fèi) EditLog 之后得到的產(chǎn)物，對(duì)于同樣的 EditLog，進(jìn)行相同順序的處理之后，始終會(huì)得到同樣的 FSDirectory。如果我們直接保存當(dāng)前的 FSDirectory 狀態(tài)，則會(huì)同消費(fèi)這些 EditLog 起到同樣的效果。在這里，CheckPoint 被看作是某個(gè)特定時(shí)間點(diǎn)下 FSDirectory 的序列化產(chǎn)物，它記錄著當(dāng)前時(shí)間點(diǎn) FSDirectory 中所有節(jié)點(diǎn)的相關(guān)信息。
HA 情況下，此時(shí)集群中存在多個(gè) NameNode 節(jié)點(diǎn)，其中只有一個(gè)是 Active 狀態(tài)，其他的都處于 Standby 狀態(tài)下。StandbyNameNode 會(huì)讀取 ActiveNameNode 的 EditLog 信息，并創(chuàng)建 CheckPoint，相關(guān)的分析會(huì)在后面的小節(jié)中討論，這里先看非 HA 模式下的實(shí)現(xiàn)。
非 HA 模式下，為了周期性對(duì) NameNode 中的文件系統(tǒng)建立 CheckPoint，通常會(huì)有一個(gè) SecondaryNameNode 伴隨 NameNode 一起啟動(dòng)，他負(fù)責(zé)周期性對(duì) NameNode 中的文件系統(tǒng)建立 CheckPoint，確保 NameNode 掛掉之后能夠快速重啟。
通過(guò) start-dfs.sh 啟動(dòng) Hdfs 集群時(shí)，如果發(fā)現(xiàn)當(dāng)前集群非 HA 集群，則會(huì)啟動(dòng)對(duì)應(yīng)的 SecondaryNameNode。
SecondaryNameNode 本身是一個(gè) Runnable 對(duì)象，在 main 方法中，會(huì)構(gòu)建一個(gè)線程類，執(zhí)行 Runnable。
while (shouldRun) {
Thread.sleep(1000 * period);
if (shouldCheckpointBasedOnCount() ||
monotonicNow >= lastCheckpointTime + 1000 * checkpointConf.getPeriod()) {
doCheckpoint();
}
}
在 SecondaryNameNode::run 中，會(huì)不斷的進(jìn)行 while 循環(huán)，當(dāng)發(fā)現(xiàn) NameNode 中執(zhí)行的 EditLog 超過(guò)一定閾值時(shí)或者距離上次 CheckPoint 的時(shí)間點(diǎn)超過(guò)閾值時(shí)，會(huì)調(diào)用 doCheckPoint 方法執(zhí)行鏡像備份操作。
// Tell the namenode to start logging transactions in a new edit file
// Returns a token that would be used to upload the merged image.
CheckpointSignature sig = namenode.rollEditLog();
RemoteEditLogManifest manifest =
namenode.getEditLogManifest(sig.mostRecentCheckpointTxId + 1);
// Fetch fsimage and edits. Reload the image if previous merge failed.
loadImage |= downloadCheckpointFiles(
fsName, checkpointImage, sig, manifest) |
checkpointImage.hasMergeError();
doMerge(sig, manifest, loadImage, checkpointImage, namesystem);
TransferFsImage.uploadImageFromStorage(fsName, conf, dstStorage,
NameNodeFile.IMAGE, txid);
doCheckPoint 中主要分為以下步驟:
- 通過(guò)
rollEditLog通知 namenode 節(jié)點(diǎn)停止對(duì)當(dāng)前 editLog 文件的寫(xiě)入操作，創(chuàng)建新的 EditLog 文件進(jìn)行 EditLog 寫(xiě)入，便于后續(xù)步驟中下載 EditLog 文件。 - 通過(guò)
getEditLogManifest獲取從最近的 CheckPoint 點(diǎn)之后的所有 EditLog 的信息，然后通過(guò)downloadCheckpointFiles從 NameNode 的 Http 服務(wù)器下載相關(guān)文件，通過(guò)doMerge進(jìn)行數(shù)據(jù)合并。 - 通過(guò)
uploadImageFromStorage將當(dāng)前合并生成了 CheckPoint Image 再傳輸回 NameNode , 作為最近一次的 CheckPoint 點(diǎn)。
之前介紹過(guò) NameNode 中有一個(gè)基于 Jetty 的 NameNodeHttpServer，負(fù)責(zé)提供 NameNode 節(jié)點(diǎn)的相關(guān)狀態(tài)，其實(shí)除了 NameNode 節(jié)點(diǎn)狀態(tài)之外，他也負(fù)責(zé)其他的 Http 請(qǐng)求訪問(wèn)，例如 SecondaryNameNode 中的下載 EditLog 和上傳 CheckPoint 文件就是通過(guò) NameNodeHttpServer::setupServlets 中注冊(cè)的 ImageServlet 類進(jìn)行實(shí)現(xiàn)的。
High Availability
在單節(jié)點(diǎn)的 NameNode 系統(tǒng)中，我們無(wú)法保證整個(gè)集群的高可用。如果 NameNode 異常退出，歸屬于這個(gè) NameNode 的所有節(jié)點(diǎn)都將不可訪問(wèn)，會(huì)導(dǎo)致整個(gè) NameNode 文件系統(tǒng)無(wú)法正常使用。為了解決這個(gè)問(wèn)題，Hdfs 中引入了 HA(High Availability) 的概念。
在 HA 環(huán)境下，會(huì)由多個(gè) NameNode 構(gòu)成一個(gè) NameService，整個(gè) NameService 中的每一個(gè) NameNode 都會(huì)存放完整的節(jié)點(diǎn)信息。通常只會(huì)有一個(gè)主節(jié)點(diǎn)(ActiveNameNode)負(fù)責(zé)向外提供服務(wù)，但是如果主節(jié)點(diǎn)異常之后，其他節(jié)點(diǎn)(StandbyNameNode)會(huì)重新進(jìn)行選主操作，選出新的主節(jié)點(diǎn)。由于每個(gè) NameNode 中都存有相對(duì)完整的節(jié)點(diǎn)信息，不會(huì)影響大多數(shù)的節(jié)點(diǎn)信息讀取。
ActiveNameNode 選舉
Hadoop 中引入了 ZooKeeper 來(lái)處理服務(wù)發(fā)現(xiàn)和節(jié)點(diǎn)異常。
通過(guò) start-dfs.sh 啟動(dòng) cluster 時(shí)，每個(gè) NameNode 節(jié)點(diǎn)上都會(huì)啟動(dòng)一個(gè) NameNode，如果判斷當(dāng)前集群是 HA 集群，則還會(huì)在每臺(tái) NameNode 的節(jié)點(diǎn)機(jī)器上啟動(dòng)一個(gè) zkfc(ZKFailoverController) 組件。
在 HA 情況下，NameNode 啟動(dòng)之后會(huì)默認(rèn)成為 StandbyNameNode，只負(fù)責(zé)同步來(lái)自 ActiveNameNode 的 EditLog。 伴隨著 NameNode 啟動(dòng)的 zkfc 負(fù)責(zé)進(jìn)行 ActiveNameNode 的選舉以及轉(zhuǎn)換 NameNode 狀態(tài)變化 。
在 zkfc 中會(huì)創(chuàng)建一個(gè) ActiveStandbyElector 處理選舉流程。 ActiveStandbyElector 會(huì)同配置 ha.zookeeper.quorum 中指定的 ZooKeeper 進(jìn)行鏈接，鏈接成功后，進(jìn)入選舉流程。
zkfc 對(duì)應(yīng)的 ZooKeeper 節(jié)點(diǎn)信息如上圖所示，hadoop-ha 是每一個(gè) NameService 注冊(cè)的頂級(jí)路徑，在它的下方是根據(jù)每個(gè) NameService 名稱命名的二級(jí)路徑，在二級(jí)路徑下是 ActiveStandbyElectorLock 和 ActiveBreadCrumb。 ActiveStandbyElectorLock 是一個(gè)臨時(shí)節(jié)點(diǎn)，它的創(chuàng)建 zkfc 被認(rèn)為是 Active。 ActiveBreadCrumb 中保存著當(dāng)前的 ActiveNameNode 信息，當(dāng) ActiveNameNode 變化時(shí)，負(fù)責(zé)通知之前的 NameNode 切換成 Standby。
具體流程圖如下:
當(dāng) zkfc 鏈接上 ZooKeeper 后，嘗試判斷 ActiveStandbyElectorLock 是否存在，若不存在，則各個(gè) zkfc 均嘗試創(chuàng)建這個(gè)臨時(shí)節(jié)點(diǎn)，通過(guò) ZooKeeper 的節(jié)點(diǎn)機(jī)制確保只有一個(gè) Client 能夠成功創(chuàng)建這個(gè)節(jié)點(diǎn)，創(chuàng)建成功的節(jié)點(diǎn)將成為主節(jié)點(diǎn)，它對(duì)應(yīng)的 NameNode 成為 ActiveNameNode,其他節(jié)點(diǎn)成為 StandbyNameNode。
SharedEditLog
在 HA 模式下，一個(gè) NameService 中只有一個(gè) ActiveNameNode 負(fù)責(zé)接收處理節(jié)點(diǎn)操作，其余的 StandbyNameNode 都只負(fù)責(zé)讀取 ActiveNameNode 中的 EditLog 數(shù)據(jù)，保證節(jié)點(diǎn)數(shù)據(jù)同 ActiveNameNode 一致。
HA 模式提供了兩種 SharedEditLog 的模式:
- NFS(Network File System): 每個(gè) NameNode 機(jī)器上如果都掛載同一個(gè) NFS，則我們可以像讀取本地文件一樣，直接從 NFS 中讀取 EditLog 信息，由于不同的 NameNode 對(duì)應(yīng)同一個(gè) NFS，因此我們讀取到的 EditLog 必然一致，只需要對(duì) EditLog 依次進(jìn)行重演，即可和 ActiveNameNode 保持同樣的狀態(tài)。 通過(guò) NFS 能夠滿足 EditLog 的同步，但是相對(duì)的，我們對(duì) NTF 的高速和無(wú)損要求也較高，因此 NTF 需要被部署在類似 NAS 的專業(yè)數(shù)據(jù)存儲(chǔ)服務(wù)器上。
- QJM(Quorum Journal Manager): 在不能保證共享磁盤(pán)的高速、無(wú)損情況下，Hdfs 提供了一個(gè) QJM 的替代方案，通過(guò)一個(gè)專門(mén)的 JournalNode 負(fù)責(zé)讀取 ActiveNameNode 上的 EditLog，StandbyNameNode 通過(guò)讀取 JournalNode 上的保留下來(lái)的 EditLog，間接實(shí)現(xiàn) EditLog 狀態(tài)同步，從而將 ActiveNameNode 上的文件讀取壓力轉(zhuǎn)移到了 JournalNode 上。
NameNode 中進(jìn)入 Standby 模式后，會(huì)啟動(dòng)一個(gè) EditLogTailer 負(fù)責(zé)從設(shè)定的 SharedEditLog 中不斷讀取 EditLog 數(shù)據(jù)。
// FSEditLog.java
private synchronized void initJournals(List<URI> dirs) {
for (URI u : dirs) {
if (u.getScheme().equals(NNStorage.LOCAL_URI_SCHEME)) {
journalSet.add(new FileJournalManager(conf, sd, storage),
required, sharedEditsDirs.contains(u));
} else {
journalSet.add(createJournal(u), required,
sharedEditsDirs.contains(u));
}
}
}
public void selectInputStreams(Collection<EditLogInputStream> streams,
long fromTxId, boolean inProgressOk, boolean onlyDurableTxns)
throws IOException {
journalSet.selectInputStreams(streams, fromTxId,
inProgressOk, onlyDurableTxns);
}
// EditLogTailer.java
void doTailEdits() throws IOException, InterruptedException {
streams = editLog.selectInputStreams(lastTxnId + 1, 0,
null, inProgressOk, true);
editsLoaded = image.loadEdits(streams, namesystem);
}
在構(gòu)建 FSEditLog 的時(shí)候，會(huì)根據(jù)配置文件中不同的 sharededitlog 配置構(gòu)造出不同的 FileJournalManager 或者 QuorumJournalManager 加入 journalSet 中。
在 StandbyNameNode 的 EditLogTailer 中，從 journalSet 中獲取 EditLogInputStream 對(duì)象，然后解析成一條條的 EditLog 進(jìn)行消費(fèi)。
CheckPoint
在非 HA 模式下，為了避免 NameNode 節(jié)點(diǎn)進(jìn)行 CheckPoint 導(dǎo)致的性能問(wèn)題，我們選擇單獨(dú)建立 SecondaryNameNode 定時(shí)進(jìn)行 CheckPoint。
在 HA 模式下， StandbyNameNode 只負(fù)責(zé)同步 EditLog 信息，對(duì)性能的要求更低，因此我們將 SecondaryNameNode 的工作分配給了 StandbyNameNode 進(jìn)行。在 StandbyNameNode 中會(huì)開(kāi)啟一個(gè) StandbyCheckPointer 線程，進(jìn)行定時(shí) CheckPoint。由于 StandbyNameNode 中已經(jīng)包含了完整的 FSDirectory 信息，不需要再?gòu)?ImageServlet 中獲取 EditLog。
