企業使用Hadoop的重大挑戰：如何在HDFS中組織和使用數據？

在上一章，我們研究了如何在MapReduce中使用不同的文件格式，以及哪些格式適合存儲數據(往期文章請查看文末鏈接)。一旦熟練掌握了數據格式的概念和使用法則，就該思考如何在HDFS中組織數據了。在設計Hadoop系統時，企業應該儘早了解如何訪問數據，以便優化將支持的重要用例，這一點非常重要。

本文作為《Hadoop從入門到精通》大型選題的第四章，主要講解影響企業數據決策的幾大因素，例如是否需要提供對數據的SQL訪問，哪些欄位用於查找數據以及訪問時間SLA。同時，企業應該確保不使用大量小文件對HDFS NameNode應用不必要的堆壓力，並且還需了解如何使用大量輸入數據集。

本章主要研究在HDFS中有效存儲和訪問大數據的方法。首先介紹在HDFS中布局數據的方法，並介紹分區和組織數據的方式以減輕NameNode堆壓力。然後，我們將討論一些數據訪問模式，以幫助處理不同的數據及龐大的數據集。最後，我們將壓縮視為一種有效方式以最大化存儲和處理數據。當然，本文只是該章的第一節，完整章節還請持續關注本專題。

本章適用於對HDFS概念有基本了解，並且具有直接使用HDFS經驗的工程師或研究人員。

4.1 組織數據

組織數據是使用Hadoop最具挑戰性的方面之一。企業中存在來自不同部門和不同人員的壓力，例如數據科學家和集群管理員，每個人對數據都有自己的要求。更重要的是，這些要求通常是在數據應用程序投入生產並且已經積累大量數據之後提出。

Hadoop中組織數據有多個維度。首先，我們需要學習如何在HDFS中組織數據，之後將面臨實際操作問題，例如對數據進行分區和壓縮，決定是否啟用Kerberos來保護集群以及管理和傳遞數據更改。本章目標是關注組織數據過程中一些複雜問題，包括數據分區和壓縮，讓我們從在HDFS中構建數據開始吧。

4.1.1 目錄和文件布局

定義數據組織方式的集群範圍標準是一項值得探究的工作，因為它可以更容易地發現數據位置，並且應用和管理可通過數據存儲解決的問題。因為我們在文件系統可以表達的範圍內工作，所以安排數據的常用方法是創建與企業組織或功能結構一致的層級結構。例如，如果在分析團隊工作並且正在將新數據集引入集群，那麼組織目錄的一種方法將如圖4.1所示。

圖4.1 HDFS目錄布局示例

在學習之前，企業最好已經確定了一種數據格式，例如Avro，這可以隨時間推移改進模式。 通過在結構中粘貼版本號，你可以靈活地轉移到任何一種新的數據格式，並使用目錄路徑傳達不同的文件格式。

在目錄中放置版本號面臨的唯一挑戰是如何將更改有效傳達給數據使用者。如果確實有該問題，企業可以嘗試將HCatalog視為從客戶端抽象出的一種數據格式。

按日期和其他欄位進行分區

你可能會使用目錄結構來模擬企業不同部門的數據演變需求，但為什麼還需要按日期進一步分區?這是Hive早期使用的一種技術，可以幫助加快查詢速度。如果將所有數據放入一個目錄中，那麼每次訪問數據實際都在進行等效的Hadoop全表掃描。相反，根據對數據的訪問方式對數據進行分區更為明智。

因為我們事先很難確切知道如何訪問數據，但按數據生成日期對數據進行分段是合理的分區嘗試。如果數據沒有日期，那麼請與數據生產者討論添加日期，因為創建事件或記錄的時間是應始終捕獲的關鍵數據點。

4.1.2 數據層

在2012年的Strata演講中，Eric Sammer提出了對數據存儲進行分層的想法，這也很好地與Nathan Marz的Lambda架構的主要原則相關——永遠不會刪除或修改原始數據。

乍一看，這似乎沒有任何意義。因為你只需要提取數據源的重要部分，其餘部分可以捨棄，畢竟保留全部原始數據有些浪費，特別是如果有部分未被積極使用，但你很難保證未來堅決不會從未使用的數據中提取到有價值的信息。

我們的軟體偶爾也會出現錯誤。想像一下，如果你正在軟體中傳輸數據，獲得想要的結果之後，你決定丟棄源數據，但是你突然發現操作邏輯中存在錯誤，因為丟棄了源數據將無法返回並重新生成結果。

因此，建議企業根據以下層級考慮數據存儲：

• 原始數據是第一層。這是從源捕獲的未更改數據，永遠不應修改此層數據，因為生成衍生物或聚合的邏輯可能存在錯誤，如果丟棄原始數據，則無法在發生錯誤時恢復。
• 第二層：從原始數據創建派生數據。該層，你可以執行重複數據刪除和任何其他數據治理要求。
• 第三層：匯總數據。這是根據派生數據計算出來的，可能會被輸入HBase系統或選擇的NoSQL系統，以便在生產和分析過程中實時訪問數據。

數據層也應該在目錄布局中有所展示，以便用戶可以輕鬆區分這些層。

一旦確定了用於對數據進行分區的目錄布局，下一步就是弄清楚如何將數據導入這些分區。

4.1.3 分區

分區是獲取數據集並將其拆分為不同部分的過程。這些部分是分區，代表了對數據的有意義劃分。數據中的公共分區示例是時間，因為它允許查詢數據的人在特定的時間窗口內縮小範圍。4.1.2節將時間作為決定在HDFS中布局數據的關鍵因素。

如果在HDFS中有一個大型數據集，需要對其進行分區，應該怎麼做呢?本節將介紹兩種可用於對數據進行分區的方法。

使用MultipleOutputs對數據進行分區

想像一下，你將股票價格數據流入HDFS，並且希望編寫MapReduce作業，以根據當天的股票報價對數據進行分區。為此，你需要在單個任務中寫入多個輸出文件，讓我們來看看如何實現這一目標。

問題

需要對數據進行分區，但大多數輸出格式僅為每個任務創建一個輸出文件。

解決方案

使用與MapReduce捆綁在一起的MultipleOutputs類。

討論

Hadoop中的MultipleOutputs類繞過了在Hadoop中生成輸出的正常通道。它提供了一個單獨的API來寫入分區輸出，並將輸出直接寫入HDFS中的任務嘗試目錄，這可以繼續提供給作業的Context對象的標準write方法來收集輸出，還可以使用MultipleOutputs來編寫分區輸出。當然，你也可以選擇僅使用MultipleOutputs類並忽略標準的基於上下文的輸出。

在此技術中，你將使用MultipleOutputs按報價日期對股票進行分區。第一步是設置MultipleOutputs以便在工作中使用。 在驅動程序中，你將指明輸出格式以及鍵和值類型：

為什麼需要在驅動程序中命名輸出?你可能想知道為什麼MultipleOutputs要求指定輸出名稱(前面示例中的分區)。這是因為MultipleOutputs支持兩種操作模式 - 靜態分區和動態分區。

如果提前知道分區名稱，靜態分區可以正常工作，這為每個分區指定不同輸出的格式提供了額外的靈活性(只需要對具有不同命名輸出的MultipleOutputs.addNamedOutput進行多次調用)。對於靜態分區，在調用addNamedOutput時指定的輸出名稱與在mapper或reducer中發出輸出時使用的名稱相同。

命名輸出主要針對於動態分區，因為在大多數情況下，操作者不會提前知道分區名稱。在這種情況下，仍然需要提供輸出名稱，當然這也可能被忽略，因為可以在mapper或reducer中動態指定分區名稱。

正如以下代碼所示，map(或reduce)類將獲取MultipleOutputs實例的句柄，然後使用write方法寫入分區輸出。請注意，第三個參數是分區名稱，即股票日期：

不要忘記close（）方法! 在任務完成後調用MultipleOutputs上的close方法非常重要。否則，輸出可能會丟失數據，甚至造成文件損壞。

正如你在以下輸出中所看到的，運行上一個示例會為單個mapper生成許多分區文件。我們還可以看到原始map輸出文件，該文件為空，因為沒有使用Context對象發出任何記錄：

此示例使用了僅map作業，但在生產中，你可能希望限制創建分區的任務數。有兩種方法可以做到這一點：

• 使用CombineFileInputFormat或自定義輸入格式來限制作業中的mapper數量。
• 使用reducer明確指定合理數量的reducer。

總結

MultipleOutputs有很多值得關注的東西：它支持「舊」和「新」MapReduce API，並支持多種輸出格式類。但是，使用MultipleOutputs應該注意一些問題：

• 在mapper中使用MultipleOutput時請記住，最終會得到NumberOfMappers * NumberOfPartition輸出文件，根據經驗，這會降低具有大量兩個值的集群!
• 每個分區在任務期間都會產生HDFS文件句柄開銷。
• 你經常會遇到大量小文件，這些文件會在分區程序的多次使用中累積。當然，這可以採用壓縮策略來緩解(有關詳細信息，請參閱第4.1.4節)。
• 儘管Avro附帶了AvroMultipleOutputs類，但由於代碼效率低下，速度很慢。

除了MultipleOutputs方法之外，Hadoop還附帶了一個MultipleOutputFormat類，其功能類似於MultipleOutputs，主要缺陷是只支持舊的MapReduce API，並且只有一種輸出格式可用於所有分區。

我們可以使用的另一種分區策略是MapReduce分區程序，它可以幫助減輕可能使用MultipleOutputs生成的大量文件。

使用自定義MapReduce分區程序

另一種分區方法是使用MapReduce中內置的分區工具。默認情況下，MapReduce使用分區程序來計算每個map輸出鍵的散列，並對reducer的數量執行計數以確定應將記錄發送到哪個reducer。我們可以編寫自定義分區程序，然後根據分區方案路由記錄來控制分區生成方式。

這種技術比以前的技術有額外的好處，因為通常會得到更少的輸出文件，每個reducer只會創建一個輸出文件，而MultipleOutputs則每個map或reduce任務都會產生N個輸出文件，每個分區一個。

問題

對輸入數據進行分區。

解決方案

編寫自定義分區程序，將記錄分區到適當的reducer。

討論

自定義分區器可向MapReduce驅動程序公開輔助方法，允許定義從日期到分區的map，並將此map寫入作業配置。然後，當MapReduce載入分區器時，MapReduce調用setConf方法; 在分區器中，我們將讀取mapping到map中，隨後在分區時使用。

驅動程序代碼需要設置自定義分區程序配置。此示例中的分區是日期，如果希望確保每個reducer對應唯一的日期。股票示例數據有10個唯一日期，因此可以使用10個reducer配置作業，還可以調用先前定義的分區幫助程序函數設置將每個唯一日期map到唯一reducer配置。

除了從輸入數據中提取股票日期並將其作為輸出key之外，mapper幾乎沒有其他功能：

運行前面示例的命令如下：

此作業將生成10個輸出文件，每個文件包含當天的股票信息。

總結

使用MapReduce框架自然地對數據進行分區可以帶來以下幾個優點：

• 對分區中的數據進行排序，因為shuffle將確保對流式傳輸到reducer的所有數據進行排序，這允許對數據使用優化的連接策略。
• 可以對reducer中的數據進行重複數據刪除，這也是shuffle階段的一個好處。

使用這種技術需要注意的主要問題是數據偏差，如果希望確保儘可能地將負載分散到reducer上，數據會存在自然偏差，這可能是一個問題。例如，如果分區是天數，那麼大多數記錄可能是一天，而可能只有一些記錄可用於前一天或後一天。 在這種情況下，理想情況是將大多數Reducer分配到一天對記錄進行分區，然後可能在前一天或後一天分配一兩個記錄，也可以對輸入進行採樣，並根據樣本數據動態確定最佳reducer數量。

生成分區輸出後，下一個挑戰是處理分區產生的大量潛在小文件。

4.1.4 數據壓縮

在HDFS中有小文件是無法避免的，也許你正在使用類似於前面描述的分區技術，或者數據可能以小文件大小有機地落在HDFS中。 無論哪種方式，都將暴露HDFS和MapReduce的一些弱點，比如：

• Hadoop的NameNode將所有HDFS元數據保留在內存中，以實現快速元數據操作。雅虎估計每個文件平均佔用內存600位元組的空間，轉換為10億個文件的元數據開銷，總計60 GB，所有文件都需要存儲在NameNode內存中。即使是當今的中端伺服器RAM容量，這也代表著單個進程的大量內存。
• 如果對MapReduce作業的輸入是大量文件，則將運行的mapper數量(假設文件是文本或可拆分的)將等於這些文件佔用的塊數。如果運行一個輸入數千或數百萬個文件的MapReduce作業，那麼將花費更多時間在內核層處理創建和銷毀map任務流程，而不是實際工作。
• 最後，如果在具有調度程序的受控環境中運行，則可能會限制MapReduce作業可以使用的任務數。由於每個文件(默認情況下)至少會生成一個map任務，因此這可能會導致作業被調度程序拒絕。

如果認為不會有這個問題，那就再想一想，文件百分比小於HDFS塊大小嗎?小多少?50%?70%?還是90%?如果大數據項目突然需要擴展以處理大幾個數量級的數據集，該怎麼辦?這不是你首先使用Hadoop的原因嗎?擴展就要添加更多節點，企業肯定不接受重新設計Hadoop並處理遷移文件。因此，在設計階段最好儘早思考和準備這些可能的問題。

本節介紹了一些可用於在HDFS中組合數據的技術。首先討論一個名為filecrush的實用程序，它可以將小文件壓縮在一起以創建較少數量的較大文件。

使用filecrush壓縮數據

壓縮是將小文件組合在一起生成更大文件的行為，這有助於緩解NameNode上的堆壓力。

就與Hadoop版本的兼容性而言，filecrush實用程序僅適用於Hadoop版本1。但是compacter(https://github.com/alexholmes/hdfscompact)以及Archive與Hadoop 版本2兼容。

問題

希望壓縮小文件以減少NameNode需要保留在內存中的元數據

解決方案

使用filecrush實用程序。

討論

filecrush實用程序組合或壓縮多個小文件以形成更大的文件。該實用程序非常複雜，並能夠

• 確定壓縮文件的大小閾值(並通過關聯，保留足夠大的文件)
• 指定壓縮文件的最大大小
• 使用不同的輸入和輸出格式以及不同的輸入和輸出壓縮編解碼器(用於移動到不同的文件格式或壓縮編解碼器)
• 使用較新的壓縮文件交換較小的文件

我們在一個簡單示例中使用filecrush ，粉碎一個小文本文件的目錄，並用gzipped SequenceFiles替換。

首先，在HDFS目錄人為創建10個輸入文件：

運行filecrush，此示例使用新的大文件替換小文件，並將文本文件轉換為壓縮的SequenceFile：

運行filecrush後會發現輸入目錄中的文件已被單個SequenceFile替換：

還可以運行文本Hadoop命令來查看SequenceFile的文本表示：

你會注意到原始的小文件已全部移動到命令中指定的輸出目錄：

如果在沒有--clone選項的情況下運行filecrush，則輸入文件將保持不變，並且已壓碎的文件將被寫入輸出目錄。

輸入和輸出文件大小閾值

filecrush如何確定文件是否需要粉碎?通過查看輸入目錄中的每個文件，並將其與塊大小進行比較(在Hadoop 2中，可以在-Ddfs.block.size命令中指定大小)。如果文件小於塊大小的75%，則會被壓縮。可以通過--threshold參數自定義此閾值，例如，如果需要將值提高到85%，則指定--threshold 0.85。

同樣，filecrush使用塊大小來確定輸出文件大小。默認情況下，它不會創建佔用超過八個塊的輸出文件，但可以使用--max-file-blocks參數進行自定義。

總結

Filecrush是一種將小文件組合在一起的簡單快捷的方法。只要存在關聯的輸入格式和輸出格式類，就支持任何類型的輸入或輸出文件。遺憾的是，它不能與Hadoop 2一起使用，並且在過去幾年中項目中沒有太多變化，因此可能企業不會考慮該程序。

因為filecrush需要輸入和輸出格式，所以如果正在處理二進位數據並且需要一種將小二進位文件組合在一起的方法，那麼它顯然還有不足之處。

使用Avro存儲多個小型二進位文件

假設正在開發一個類似於Google圖像的項目，可以在其中抓取網頁並從網站下載圖像文件，該項目是互聯網規模，因此下載了數百萬個文件並將它們單獨存儲在HDFS中。你已經知道HDFS不能很好地處理大量小文件，因此我們嘗試一下Avro。

問題

希望在HDFS中存儲大量二進位文件，並且無需遵守NameNode內存限制即可。

解決方案

在HDFS中處理小型二進位文件的最簡單方法是將其打包到一個更大的文件中。此技術將讀取存儲在本地磁碟目錄中的所有文件，並將它們保存在HDFS的單個Avro文件中。你還將了解如何使用MapReduce中的Avro文件來處理原始文件內容。

討論

圖4.2顯示了該技術的第一部分，可以創建HDFS中的Avro文件。這樣做可以在HDFS中創建更少的文件，這意味著存儲在NameNode內存中的數據更少，這也意味著可以存儲更多內容。

圖4.2 在Avro中存儲小文件以允許存儲更多文件

Avro是由Hadoop的創建者Doug Cutting發明的數據序列化和RPC庫。Avro具有強大的架構演進功能，其優勢已經超過競爭對手，比如SequenceFile等，第3章詳細介紹了這一內容，此處不再贅述。

請查看以下Java代碼，該代碼將創建Avro文件：

代碼4.1讀取包含小文件的目錄，並在HDFS中生成單個Avro文件

要運行本章代碼需要在主機上安裝Snappy和LZOP壓縮編解碼器。有關如何安裝和配置它們的詳細信息，此處不作贅述。

當針對Hadoop的config目錄運行此腳本時會發生什麼(將$ HADOOP_CONF_DIR替換為包含Hadoop配置文件的目錄)：

讓我們確保輸出文件是HDFS：

為確保一切正常運行，還可以編寫代碼以從HDFS讀取Avro文件，並為每個文件的內容輸出MD5哈希值：

此代碼比寫入更簡單。由於Avro將架構寫入每個Avro文件，因此無需在反序列化過程中向Avro提供有關架構的信息：

此時，我們已經在HDFS中擁有了Avro文件。儘管本章是關於HDFS的，但可能要做的下一件事是處理在MapReduce中編寫的文件。這需要編寫一個Map-only MapReduce作業，它可以讀取Avro記錄作為輸入，並輸出一個包含文件名和文件內容的MD5哈希值的文本文件，如圖4.3所示。

圖4.3 map作業讀取Avro文件並輸出文本文件

以下顯示了此MapReduce作業的代碼：

代碼4.2 一個MapReduce作業，包含小文件的Avro文件作為輸入

如果通過之前創建的Avro文件運行此MapReduce作業，則作業日誌文件將包含文件名和哈希：

該技術假設使用的文件格式(例如圖像文件)無法將單獨的文件連接在一起。如果文件可以連接，應該考慮該選項。採用該方法請盡量確保文件大小與HDFS塊一樣，以最大限度減少存儲在NameNode中的數據。

總結

我們可以使用Hadoop的SequenceFile作為保存小文件的機制。SequenceFile是一種比較成熟的技術，比Avro文件更成熟。但是SequenceFiles是特定於Java的，並且不提供Avro帶來的豐富的互操作性和版本控制語義。

谷歌的Protocol Buffers以及Apache Thrift(源自Facebook)也可用於存儲小文件。 但是都沒有適用於本機Thrift或Protocol Buffers文件的輸入格式。我們也可以使用另一種方法就是將文件寫入zip文件。缺點是必須編寫自定義輸入格式來處理zip文件，其次是zip文件不可拆分(與Avro文件和SequenceFiles相反)。這可以通過生成多個zip文件並嘗試使它們接近HDFS塊大小來減輕。Hadoop還有一個CombineFileInputFormat，可以將多個輸入拆分(跨多個文件)提供給單個map任務，這大大減少了運行所需的map任務數量。

我們還可以創建一個包含所有文件的tarball文件，生成一個單獨的文本文件，其中包含HDFS中tarball文件的位置。此文本文件將作為MapReduce作業的輸入提供，mapper將直接打開tarball。但是，這種方法會繞過MapReduce的局部性，因為mapper將被安排在包含文本文件的節點上執行，因此可能需要從遠程HDFS節點讀取tarball塊，從而導致不必要的網路I/O。

Hadoop歸檔文件(HAR)是專門為解決小文件問題而創建的，是位於HDFS之上的虛擬文件系統。HAR文件的缺點是無法針對MapReduce中的本地磁碟訪問進行優化，並且無法進行壓縮。Hadoop版本2支持HDFS Federation，其中HDFS被劃分為多個不同的命名空間，每個命名空間由一個單獨的NameNode獨立管理。實際上，這意味著將塊信息保存在內存中的總體影響可以分布在多個NameNode上，從而支持更多數量的小文件。

最後，提供Hadoop發行版的MapR擁有自己的分散式文件系統，支持大量小文件。將MapR用於分散式存儲對系統來說需要很大改變，因此，企業不太可能轉移到MapR來緩解HDFS的這個問題。你可能會遇到想要在Hadoop中處理小文件的時間，並且直接使用它們會導致膨脹的NameNode內存和運行緩慢的MapReduce作業。此技術通過將小文件打包到更大的容器文件中來幫助緩解這些問題。我之所以選擇Avro是因為它支持可拆分文件，壓縮及其富有表現力的架構語言，這將有助於版本控制。如果文件很大，想要更有效地存儲應該怎麼辦呢?這將在第4.2節得到解決。

4.1.5原子數據移動

分區和壓縮等行為往往遵循類似的模式，比如在暫存目錄中生成輸出文件，然後需要在成功暫存所有輸出文件後以原子方式移動到最終目標目錄。這可能會帶來一些問題：

• 使用什麼觸發器來確定已準備好執行原子移動?
• 如何在HDFS中原子移動數據?
• 數據移動對最終數據讀者有何影響?

使用MapReduce驅動程序作為後處理步驟執行原子移動可能很好，但如果客戶端進程在MapReduce應用程序完成之前死亡會發生什麼?這是在Hadoop中使用OutputCommitter非常有用的地方，因為可以將任何原子文件移動作為工作的一部分，而不是使用驅動程序。

接下來的問題是如何在HDFS中原子移動數據。在最長的時間內，人們認為DistributedFileSystem類(這是支持HDFS的具體實現)上的重命名方法是原子的。但事實證明，在某些情況下，這不是原子操作。這在HADOOP-6240中得到了解決，但出於向後兼容性原因，重命名方法未更新。因此，重命名方法仍然不是真正的原子。相反，你需要使用新的API。如下所示，代碼很麻煩，只適用於較新版本的Hadoop：

HDFS缺少的是原子交換目錄能力，這在壓縮等情況下非常有用。因為在這些情況下，你需要替換其他進程(如Hive)正在使用的目錄的全部內容。有一個名為「「Atomic Directory Swapping Operation」的項目可能有對你有些幫助。

以上就是使用Hadoop系統需要掌握的數據組織技術及相應原則。下一節我們將介紹Hadoop中的另一個重要數據管理主題——數據壓縮。

喜歡這篇文章嗎？立刻分享出去讓更多人知道吧！