成功运行的部落格

1.  背景介紹
談到分布式系統，就不得不提到Google的三駕馬車:GFS[1],MapReduce[2]和BigTable[3]。雖然Google沒有開源這三個技術的實現源碼，但是基於這三篇開源文檔, Nutch項目子項目之一的Yahoo資助的Hadoop分別實現了三個強有力的開源產品：HDFS，MapReduce和HBase。在大數據時代的背景下，許多公司都開始采用Hadoop作為底層分布式系統，而Hadoop的開源社區日益活躍，Hadoop家族不斷發展壯大，已成為IT屆最炙手可熱的產品。
本文將在簡單介紹Hadoop主要成員的基礎上，探討Hadoop在應用中的改進。
第一部分是對Hadoop誕生和現狀的簡單描述。
第二部分將簡單介紹hadoop的主要成員，主要包括他們的基本特性和優勢。分別是分布式文件系統HDFS，NoSQL家族之一的HBase，分布式並行編程方式MapReduce以及分布式協調器Zookeeper。
第三、四、五部分分別介紹了Hadoop的不同改進和使用。按次序分別是facebook的實時化改進，HadoopDB，以及CoHadoop。
最後是我的總結和體會。
如果對Hadoop的基本架構和基礎知識熟悉，可以從第三部分看起。
2.  關於Hadoop
Hadoop本身起源於Apache Nutch項目，曾也是Lucene項目的一部分。從結構化數據，到半結構化數據和非結構化數據，從關系型數據庫到非結構化數據庫（NoSQL），更高性能的並行計算/批處理能力和海量數據存儲成為現代主流IT公司的一致需求。
2.1  HDFS
HDFS，全稱Hadoop Distributed Filesystem，是Hadoop生態圈的分布式文件系統。分布式文件系統跨多臺計算機存儲文件，該系統架構於網絡之上，誕生即具備了網絡編程的復雜性，比普通磁盤文件系統更加復雜。
2.1.1  HDFS數據塊
HDFS以流式數據訪問模式來存儲超大文件，運行於商用硬件集群上。數據集通常由數據源生成或從數據源復制而來，接著長時間在此數據集上進行格類分析處理。每次都將涉及該數據集的大部分數據甚至全部，因此讀取整個數據集的時間延遲比讀取第一條記錄時間的延遲更重要。而一次寫入、多次讀取是最高效的訪問模式。有一點要說明的是，HDFS是為高數據吞吐量應用優化的，而這可能會以高時間延遲為代價。
HDFS默認的最基本的存儲單元是64M的數據塊(block)。HDFS的塊比磁盤塊(512字節)大得多，目的是為了最小化尋址開銷。HDFS上的文件也被劃分為多個分塊(chunk)，作為獨立存儲單元。與其他文件系統不同的是，HDFS中小於一個塊大小的文件不會占據整個塊的空間。
塊抽象給分布式文件系統帶來的好處：
 ?  文件的大小可以大於網絡中任意一個磁盤的容量。
 ?  使用塊抽象而非整個文件作為存儲單元，大大簡化了存儲子系統的設計，同時也消除了對元數據的顧慮。
?  塊非常適合用於數據備份進而提供數據容錯能力和可用性。
2.1.2   Namenode和Datanode
namenode和datanode的管理者-工作者模式有點類似主從架構。namenode對應多個datanode。namenode管理文件系統的命名空間，維護文件系統和內部的文件及目錄。datanode是文件系統的真正工作節點，根據需要存儲並檢索數據塊（一般受namenode調度），並且定期向namenode發送它們所存儲的塊的列表。
namenode一旦掛掉，文件系統的所有文件就丟失了，不知道如何根據datanode的塊來重建文件。因此，namenode的容錯或者備份是很重要的。在HDFS中存在secondarynamenode（雖然不完全是個namenode的備份，更確切的是個輔助節點）周期性將元數據節點的命名控件鏡像文件和修改日誌合並。
2.2  HBase
跟傳統的關系型數據庫（RDBMS）基於行存儲不同，HBase是一個分布式的，在HDFS上開發的面向列的分布式數據庫。HBase行中的列分成“列族”（column family），所有的列族成員有相同的前綴。所有列族成員都一起存放在文件系統中。
2.2.1   與RDBMS比較
HBase通過在HDFS上提供隨機讀寫來解決Hadoop不能處理的問題。HBase自底層設計開始即聚焦於各種可伸縮性問題：表可以很“高”，有數十億個數據行；也可以很“寬”，有數百萬個列；水平分區並在上千個普通商用機節點上自動復制。表的模式是物理存儲的直接反映，使系統有可能提高高效的數據結構的序列化、存儲和檢索。
而RDBMS是模式固定、面向行的數據庫且具有ACID性質和復雜的SQL查詢處理引擎，強調事物的強一致性（strong consistency）、參照完整性（referential integrity）、數據抽象與物理存儲層相對獨立，以及基於SQL語言的復雜查詢支持。
2.2.2   HBase特性
簡單列舉下HBase的關鍵特性。
?  沒有真正的索引：行是順序存儲的，每行中的列也是，所以不存在索引膨脹的問題，而且插入性能和表的大小有關。
?  自動分區：在表增長的時候，表會自動分裂成區域（region），並分布到可用的節點上。
?  線性擴展：對於新增加的節點，區域自動重新進行平衡，負載會均勻分布。
?  容錯：大量的節點意味著每個節點重要性並不突出，所以不用擔心節點失效問題。
?  批處理：與MapReduce的集成可以全並行地進行分布式作業。
2.3  MapReduce
MapReduce是一種可用於數據處理的編程模型，是一個簡單易用的軟件框架，基於它寫出來的應用程序能夠運行在由上千個商用機器組成的大型集群上，並以一種可靠容錯的方式並行處理上T級別的數據集。
2.3.1 Map & Reduce
一個Map/Reduce 作業（job）通常會把輸入的數據集切分為若幹獨立的數據塊，由 map任務以完全並行的方式處理它們。框架會對map的輸出先進行排序，然後把結果輸入給reduce任務。通常作業的輸入和輸出都會被存儲在文件系統（一般為HDFS）中。整個框架負責任務的調度和監控（jobtracker協調作業的運作，tasktracker運行作業劃分後的任務），以及重新執行已經失敗的任務。
通常，Map/Reduce框架和分布式文件系統是運行在一組相同的節點上的，也就是說，計算節點和存儲節點通常在一起。這種配置允許框架在那些已經存好數據的節點上高效地調度任務，這可以使整個集群的網絡帶寬被非常高效地利用。
2.3.2 Matser/Slave架構
Map/Reduce框架由一個單獨的master JobTracker 和每個集群節點一個slave TaskTracker共同組成。master負責調度構成一個作業的所有任務，這些任務分布在不同的slave上，master監控它們的執行，重新執行已經失敗的任務。而slave僅負責執行由master指派的任務。
應用程序至少應該指明輸入/輸出的位置（路徑），並通過實現合適的接口或抽象類提供map和reduce函數。再加上其他作業的參數，就構成了作業配置（jobconfiguration）。然後，Hadoop的 job client提交作業（jar包/可執行程序等）和配置信息給JobTracker，後者負責分發這些軟件和配置信息給slave、調度任務並監控它們的執行，同時提供狀態和診斷信息給job-client。
2.4  Zookeeper
Zookeeper是一個高可用的分布式數據管理與系統協調框架。簡單的說，就是個分布式協調器。它以主從的架構，基於Paxos算法實現，保證了分布式環境中數據的強一致性，也因此各種分布式開源項目中都有它的身影。
2.4.1  Zookeeper機制
Zookeeper的核心是一個精簡的文件系統，它的原語操作是一組豐富的構件（building block），可用於實現很多協調數據結構和協議，包括分布式隊列、分布式鎖和一組同級節點中的“領導者選舉”（leader election）。
Zookeeper實現的是Paxos算法。Zookeeper集群啟動後自動進行leader selection，投票選出一臺機器作為Leader，其他的都是Follower。通過heartbeat的機制，Follower從Leader獲取命令或者消息，同步自己的數據，和Leader保持一致。為了保證數據的一致性，只有當半數以上的Follower的狀態和Leader成功同步了之後，才認為這次數據更新是成功的。為了選舉方便，Zookeeper集群數目是奇數。
3. Hadoop在Facebook變得實時[4]
論文主要解釋了Facebook引進Hadoop的原因。結合自己的需求，Facebook對hadoop進行了更實時的改進。
3.1  HDFS與MySQL的性能互補
HDFS適合大塊地讀取數據（推薦節點是64M），它關於隨機讀取的工作的accesslatency比較大，所以一般會用大規模的MySQL集群結合memcached這樣的緩存工具來做處理。在Facebook中，從Hadoop中產生的類似中間結果的數據會裝載到MySQL集群或者memcached中去，用來被web層使用。
同時，HDFS的順序讀取性能很好。Facebook需求寫方面的高吞吐量，代價低的彈性存儲，同時要求低延遲和硬盤上高效的順序和隨機讀取。MySQL存儲引擎被證明有比較高的隨機讀取能力，但是隨機寫吞吐率比較差。因此，Facebook決定采用Hadoop和HBase來平衡順序和隨機讀取的性能，而不是只采用MySQL集群來不斷嘗試一種難以把握的balance。具體Facebook的需求將在下一節仔細剖析。
3.2  Facebook需求
Facebook認為，用他們已有的基於MySQL集群的一些解決方案來處理問題已經遇到了瓶頸。之前的用例對工作量的擴展是有挑戰性的。在一個RDBMS的環境下解決非常高的寫吞吐量，大數據，不可預測增長及其他問題變得十分困難。
3.3  選擇Hadoop和HBase原因
采用Hadoop和HBase來解決以上需求的存儲系統方案的原因可以總結為以下幾點：
?  彈性：需要能夠用最小的開銷和零宕機修復時間來對存儲系統增量式地擴容。這裏的擴容應該指的是可以比較方便地實時增加服務器臺數來應對一些高峰或者突發服務需求。
?  高的寫吞吐量
?  高效的硬盤隨機讀寫
?  高可用性和容災
?  錯誤隔離：當局部數據庫掛掉或者服務器不能提供服務的時候，讓最少的用戶受到影響。HDFS應對這樣的場景還是很不錯的。
?  讀寫改的原子性：底層存儲系統針對高並發量的需求
?  範圍掃描：指特定場景下高效獲取一個範圍結果集。
HBase已經以key-value存儲的方式提供了高一致性的高寫吞吐，且在大規模數據傳送和快速隨機寫以及流式讀方面表現優異。它同時保證了行層次的原子性。從數據模型的角度看，面向列的實現給數據存儲帶來了極高的靈活性，“寬”行允許在一個table內存放百萬數量級的被索引的值。
雖然HDFS的核心namenode的宕機會帶來巨大影響，但是Facebook有信心打造一個在合理時限內的高可用的NameNode。根據一些實踐測試，Facebook對HDFS進行了設計和改進，主要針對namenode。將在下節展開。
3.4  實時HDFS
HDFS剛開始是為了支持MapReduce這樣的並行應用的數據存取的，是面向批處理系統的，所以在實時方面講本身可能是存在不足的。Facebook主要改造在於一個高可用的AvatarNode。
我們知道HDFS的namenode一旦掛掉，整個集群就得等到namenode再次啟動才能繼續運行提供服務，所以需要這個熱備份——AvatarNode的設計。在HDFS啟動的時候，namenode是從一個叫fsimage的文件裏讀取文件系統的元數據的。元數據信息包括了HDFS上所有文件和目錄的名字和元數據。但是namenode不會持續地去存每一塊block的位置信息。所以冷啟動namenode的時候包括兩部分：首先讀文件系統鏡像；然後，大部分datanode匯報進程上的block信息，以此來恢復集群上每一塊已知block的位置信息。這樣的冷啟動會花很長時間。
雖然一個備用的可用node可以避免failover時候去讀磁盤上的fsimage，但是依然需要從datanodes裏獲取block信息。所以，時間相對還是偏長。於是誕生了AvatarNode。


HDFS擁有兩個AvatarNode——Active AvatarNode和Standby AvatarNode。他們形成了一對“主被動熱備份”（active-passive-hot-standby）。AvatarNode是對NameNode的包裝。Facebook的HDFS集群都采用NFS來存一份文件系統鏡像的備份和一份事物日誌的備份。Active AvatarNode把自己處理的事務寫進NFS裏的事務日誌。同時，StandbyAvatarNode打開NFS上同一份事務日誌，然後在自己的命名空間內開始執行事務，以保證自己的命名空間盡可能和初始信息接近。Standby AvatarNode同時照顧到初始信息的核查並創建新的文件系統鏡像，和HDFS相比就沒有了分離的SecondNameNode。
Datanodes同時和兩個AvatarNode交流。這保證了Standby處也獲得到最新的block狀態信息，以在分鐘時間級內轉化成為Activer的Node（之前說namenode的冷啟動的時長問題可以解決了）。Avatar DataNode相互之間輸送心跳，block信息匯報和接受到的block。Avatar DataNodes集成了Zookeeper，因此他們知道主節點信息，會執行主節點發送的復制/刪除命令（基於Zookeeper的leader selection和heartbeat機制），而來自Standby AvatarNode的復制/刪除請求是忽略的。
對於事務日誌的記錄，還進行了一些改進。
i.   為了讓故障和失效盡可能透明，Standby必須知道失效發生時的block位置信息，所以對每一塊block分配記錄一個額外的記錄日誌。這樣允許客戶端在發生失效的時刻前還是一直在寫文件。
ii.  當Standby向正在被Active寫事務記錄的日誌裏讀取事務信息的時候，有可能讀到的是一個局部的事務。為了避免這樣的問題，給每個要寫進日誌裏的事務增加記錄事務長度信息，事務id和校驗和。
要了解更具體的信息，可以從原paper中獲得更多具體的情況。
4.  HadoopDB[6]
HadoopDB簡單介紹下設計理念和他的架構。
4.1 HadoopDB理念
HadoopDB是一個混合系統。基本思想是用MapReduce作為與正在運行著單節點DBMS實例的多樣化節點的通信層。查詢語言用SQL表示，並用現有工具翻譯成MapReduce可以接受的語言，使得盡可能多的任務可以被推送到每個高性能的單節點數據庫上。這樣基於MapReduce的並行化的數據庫代價幾乎是零。因為MapReduce是現有的。
HadoopDB背後的一些主要思想包括以下兩個關鍵字：share-nothing MPP架構和parallel databases。
4.2 HadoopDB架構介紹


作為一個混合的系統，讓我們看看HadoopDB由哪些部分構成：HDFS，MapReduce，SMS Planner，DB Connector等等。HadoopDB的核心框架還是Hadoop，具體就是存儲層HDFS，和處理層MapReduce。關於HDFS上namenode，datanode各自處理任務，數據備份存儲機制以及MapReduce內master-slave架構，jobtracker和tasktracker各自的工作機制和任務負載分配，數據本地化特性等內容就不詳細說了。下面對主要構成部件做簡單介紹：
1.    Databae Connector：承擔的是node上獨立數據庫系統和TaskTracker之間的接口。每個single的數據庫都關聯一個datanode和一個tasktracker。他傳輸SQL語句，得到一些KV返回值。擴展了Hadoop的InputFormat，使得與MapReduce框架實現無縫拼接。
2.    Catalog：維持數據庫的元數據信息。包括兩部分：數據庫的連接參數和元數據，如集群中的數據集，復本位置，數據分區屬性。現在是以XML來記錄這些元數據信息的。由JobTracker和TaskTracker在必要的時候來獲取相應信息。
3.    Data Loader：主要職責涉及根據給定的分區key來裝載數據，對數據進行分區。包含自身兩個主要Hasher：Global Hasher和Local Hasher。簡單地說，Hasher無非是為了讓分區更加均衡。
4.    SMS Planner：SMS是SQL to MapReduce to SQL的縮寫。HadoopDB通過使他們能執行SQL請求來提供一個並行化數據庫前端做數據處理。SMS是擴展了Hive。關於Hive我在這裏不展開介紹了。總之是關於一種融入到MapReduce job內的SQL的變種語言，來連接HDFS內存放文件的table。

5.  CoHadoop[7]
論文提出CoHadoop來解決Hadoop無法把相關的數據定位到同一個node集合下的性能瓶頸。CoHadoop是對Hadoop的一個輕量級擴展，目的是允許應用層能控制數據的存儲。應用層通過某種方式提示CoHadoop某些集合裏的文件是相關性比較大的，可能需要合並，之後CoHadoop就嘗試去轉移這些文件以提高一定的數據讀取效率。
5.1  研究意義
Hadoop++[6]項目其實也做過類似的事，它將同一個job產生的兩個file共同放置，但是當有新文件註入系統的時候，它需要對數據重新組織。
CoHadoop的改進主要給以下幾個操作帶來了比較大的好處：索引（indexing），聚合（grouping），聚集（aggregation），縱向存儲（columnar storage），合並（join）以及sessionization。而像日誌分析這樣的操作，涉及到的就是把一些參考數據合並起來或者進行sessionization。這可以體現CoHadoop的改進意義所在。
以下是paper關於CoHadoop的總結：
?  這是一種很靈活，動態，輕量級的共置相關數據文件的方案，而且是直接在HDFS上實現的。
?  在日誌處理方面，確定了兩個用例：join和sessionization，使得在查詢處理方面得到了顯著的性能提高。
?  作者還研究了CoHadoop的容錯，分布式數據和數據丟失。
?  在不同的場景下測試了join和sessionization的效果。
接下來還是介紹下CoHadoop的設計思想。
5.2  改進設計介紹
HDFS本身存數據的時候是有冗余的。默認是存三分拷貝。這三份復制品會存在不同的地方。最簡單是存在datanode裏。默認的存放方式是第一份拷貝存在新建的本地誕生的node的block裏（假設足夠存），這叫寫“親和”（write affinity）。HDFS然後選擇同一機架上的datanode存放第二個拷貝，選擇不同機架上的一個datanode存第三份拷貝。這是HDFS的本來的機制。那麽為了實現相關數據的共置存儲，論文修改了存放策略。
以上Hadoop現有的存放策略主要是為了負載均衡，但是當應用需要從不同的文件裏去取所需的數據的時候，如果能自定義一些策略，那可能會得到顯著的提升。輕量級的CoHadoop使得開發自定義的策略變得簡單。雖然分區在Hadoop裏實現很簡單，但是共置並不容易，Hadoop也沒有提供這樣類似的可行性功能實現。

CoHadoop擴展了HDFS，提出了新的文件層屬性——locator，並且修改了Hadoop的數據存放策略以使用這個locator。假設每個locator由一個整數值表示（也可以是別的表示方法），那麽文件和locator之間可以是一個N：1的關系。每個HDFS的文件最多和一個locator關聯，同一個locator可以關聯很多文件。同一個locator下的文件存在同一個datanode集合裏，而沒有locator映射的文件依舊按照默認的Hadoop的存儲機制存放。
為了更好地管理和跟蹤這些locator和文件之間的映射信息，設計了一個新的數據結構——locatortable存在namenode裏。它存放了每個locator映射的文件集。當namenode運行的時候，locator table是在內存裏動態維護的，
關於數據存放策略的修改是這麽做的：只要有一個新的和locator l關聯的文件f被創建，會去locator table裏查詢是否存在一個實例是屬於這個locator l的。如果不存在，就新增一條（l, f）在table裏，並用HDFS默認的存放方式存這份文件的拷貝們。如果已經存在，就可以知道這個l映射的file list，如果從現有的存放了這個list內的文件的r個datanode裏按一定方式（考慮空間）選出幾個用於存新來的文件的拷貝的節點，存放這份文件的拷貝們。大致的意思就是這樣。
關於日誌的join和sessionization的改進，就不展開了。


做sessionization，對於日誌處理時候MapReduce計算的影響比較。
6.  總結
雖然我對Hadoop有濃厚的興趣，但是自己所能接觸到的項目和環境，都沒有到達一個比較飽和的需求點。要做分布式存儲？根本用不著動用HBase或者別的NoSQL組成的分布式集群，只需要一個分布式的MySQL集群就可以了，NoSQL可以做的事，其實MySQL何嘗不能完成？只是說NoSQL對某些數據的存儲，在某些讀寫性能上有局部的個性化的優勢而已。更不必說要用MapReduce去完成什麽樣大規模，TB級數據的分布式並行計算了。在數據和硬件設施方面，以至到技術程度方面，學校裏都沒有滿足條件，沒有如此的需求。
學校的課程裏也沒有涉及到分布式的內容。分布式文件系統/存儲/索引之類的話題一直是存在於企業級別，存在於大公司大數據基礎和服務器集群基礎的。學校裏偶爾可以聽到如阿裏開的關於分布式的講座，也是很基礎的，淺嘗截止。
出生在什麽樣的年代，就會接觸什麽樣的技術。學習什麽樣的技術，就能充實自己成什麽樣的技術人才。把握Hadoop，把握時代的核心技術，就掌握了現在大數據時代，甚至可以遇見並操控未來！