MapReduce如何實現Reduce端重分區Join操作優化

MapReduce如何實現Reduce端重分區Join操作優化，很多新手對此不是很清楚，為了幫助大家解決這個難題，下面小編將為大家詳細講解，有這方面需求的人可以來學習下，希望你能有所收獲。

一、重分區Join操作（Reduce端）

本文介紹的第一種方法是最基本的重分區Join操作，該方法允許執行內部和外部Join。開始之前，我們先搞清楚要解決的問題是將大型數據集Join在一起，我們選用的解決方案是Reduce端重分區Join。該方法是一種Reduce端Join實現，利用MapReduce的sortmerge將記錄組合在一起，作為單個MapReduce作業實現，可支持N路連接，其中N是要連接的數據集數量。

Map端負責從數據集中讀取數據，確定每個Join操作的value，并將該value的key輸出，輸出key包含在reducer中并將數據集組合在一起以生成最終結果。

單個reducer調用接收map函數Join操作發出的Key對應的所有值，并將數據分N個分區，其中N是要連接的數據集數量。reducer讀取連接value的所有輸入并將它們分區到內存中，然后跨所有分區執行笛卡爾積，并發出每個Join操作的結果。

圖6.10 重分區Join操作的基本MapReduce實現

MapReduce代碼要支持這種技術，需要滿足以下條件：

• 支持多個map類，每個map類處理不同的輸入數據集，這是通過使用MultipleInputs類完成的。

• 需要一種方法來標記mapper發出的記錄，以便可以與其原點的數據集相關聯，本文將使用htuple項目處理MapReduce中的數據。

重分區Join操作的代碼如下：

可以使用以下命令運行作業并查看輸出：

總結

Hadoop捆綁了一個hadoop-datajoin模塊，這是一個重分區Join操作框架，包括用于處理多個輸入數據集和執行Join操作的管道。上述操作示例及hadoop-datajoin代碼是重分區Join的最基本形式，兩者都要求在執行笛卡爾積之前將連接key的所有數據加載到內存中，但如果連接key的基數大于可用內存，那么，這種方法就不太適用。下一個技術將著眼解決此問題。

二、優化重分區Join操作

舊版重分區Join操作實現會浪費大量空間，需要將給定key的所有value加載到內存中才能執行多路連接，將較小的數據集加載到內存中才能迭代更大的數據集，沿途執行Join更有效。

我們希望在MapReduce中執行重分區Join，且無需緩存reducer中的所有記錄。優化后的重分區Join框架將僅緩存要連接的其中一個數據集，以減少reducer中緩存的數據量。此優化僅緩存來自兩個數據集中較小者的記錄，以減少緩存所有記錄的內存開銷，圖6.11顯示了改進的重分區Join實現。

圖6.11 重分區Join操作優化MapReduce實現

該技術與舊版相比存在一定差異，此處使用輔助排序確保來自較小數據集的所有記錄在較大數據集的記錄之前到達reducer，以此來盡可能減少reducer中要緩存的數據量。此外，mapper會發出需要進行Join操作的用戶名元組的key以及標識原始數據集的字段。

以下代碼顯示了一個新的枚舉，顯示了用戶mapper如何填充元組字段：

需要更新MapReduce驅動程序代碼以指示元組中的哪些字段應用于排序、分區和分組：

• 分區程序應僅基于用戶名進行分區，以便用戶的所有記錄都到達同一個reducer。

• 排序應使用用戶名和數據集指示符，以便首先排序較小的數據集(由于USERS常量小于USER_LOGS常量，導致用戶記錄在用戶登錄之前排序)。

• 分組應對用戶進行分組，以便將兩個數據集都流式傳輸到同一個reducer調用：

最后，我們要修改reducer以緩存傳入的用戶記錄，然后將其與用戶日志Join：

可以使用以下命令來運行作業并查看輸出：

Hive

在執行重分區Join操作時，Hive可支持類似優化。Hive可緩存Join鍵的所有數據集，然后流式傳輸大型數據集，使其不需要存儲在內存中。假定在查詢時，Hive最后指定的數據集最大。想象一下，你有兩個名為users和user_logs的表，而user_logs要大得多。要連接這些表，我們需要確保user_logs表被引用為查詢中的最后一個：

如果不想重新查詢，可以使用STREAMTABLE提示告訴Hive哪個表更大：

總結

此操作實現通過僅緩沖較小數據集的value來改進早期技術，但它仍然存在數據在map和reducer之間的傳輸問題，這是一個昂貴的網絡成本。此外，舊版可以支持N路連接，但是這種實現僅支持雙向連接。

三、使用Bloom過濾器來減少混洗數據

如果希望根據某些謂詞對數據子集執行Join操作，例如“僅限居住在加利福尼亞地區的用戶”。到目前為止，我們還必須在reducer中執行過濾器才可以實現這一目的 ，因為只有一個數據集存放了有關狀態的詳細信息——用戶日志沒有該信息。接下來，我將介紹如何在map端使用Bloom過濾器，這會對作業執行時間產生很大影響。我要解決的問題是在重分區Join操作中過濾數據，但要將該過濾器推送到mapper。一個可行的解決方案是使用預處理作業創建Bloom過濾器，然后在重分區作業中加載Bloom過濾器以過濾mapper中的記錄。

Bloom過濾器是一種非常有用的隨機數據結構，它利用位數組簡潔表明集合，并能判斷一個元素是否屬于該集合。然而，與Java中的HashSet相比，Bloom需要的內存要少得多，因此它們非常適合處理大型數據集。此解決方案有兩個步驟，一是運行作業來生成Bloom過濾器，該過濾器將對用戶數據進行操作，并由居住在加利福尼亞地區的用戶填充；二是在重分區Join操作中使用此Bloom過濾器丟棄不需要的用戶，該過程需要Bloom過濾器的原因是用戶日志的mapper沒有狀態的詳細信息。

圖6.12 在重分區Join中使用Bloom過濾器的兩步過程

第1步：創建Bloom過濾器

第一個作業是創建Bloom過濾器，其中包含加利福尼亞州的用戶名。mapper生成中間Bloom過濾器，reducer將其組合成一個Bloom過濾器，作業輸出是包含序列化Bloom過濾器的Avro文件：

第2步：重分區Join

重分區Join與上文提到的唯一區別是mapper加載第一步中生成的Bloom過濾器，并且在處理map記錄時，執行針對Bloom過濾器的元素審查以確定是否應將記錄發送給reducer。以下代碼顯示了兩件事：一般化Bloom過濾器加載、抽象mapper以及支持兩個Join數據集的子類：

以下命令運行兩個作業并轉儲Join輸出：

總結

該技術提出了一種在兩個數據集上執行map端過濾的有效方法，以最小化mapper和reducer之間的網絡I/O。作為shuffle的一部分，它還減少了mapper和reducer的磁盤溢出數據量。過濾器通常是加速和優化作業最簡單有效的方法，重分區Join也同樣適用于其他MapReduce作業。

四、reducer端Join操作可能發生數據傾斜

數據傾斜是實際操作中很容易碰到的問題，可能存在兩種類型的數據傾斜：

• 高Join-key基數，其中有一些連接key在一個或兩個數據集中具有大量記錄，我把這種稱之為join-product偏差。

• 糟糕的散列分區，少數reducer在總記錄數中占很大比例，我將此稱為散列分區傾斜。

五、加入具有高連接密鑰基數的大型數據集

這種技術解決了join-product的傾斜問題，下一個技術檢查了散列分區偏差。現在面臨的問題是某些連接key是高基數的，這會導致某些reducer在嘗試緩存這些key時耗盡內存。我們可以過濾掉這些key并將它們單獨連接或將其溢出到reducer中并安排后續作業Join。

如果提前知道了哪些Key是高基數的，則可以將其分成單獨的Join作業，如果不確定高基數Key是哪些，則可能需要在reducer中構建智能檢測并將其寫入副本文件，該文件由后續作業Join，如圖6.14所示。

圖6.13 提前知道高基數密鑰時處理傾斜

圖6.14 提前知道高基數密鑰處理時的偏差

Hive

Hive支持類似于第二種方法的偏斜緩解策略，運行作業之前可指定以下配置啟用：

可以選擇設置一些其他配置來控制在高基數key上運行的map端連接：

最后，如果在SQL中使用GROUP BY，可能還需要考慮啟用以下配置來處理分組數據中的偏差：

總結

此技術假設給定的Join鍵，只有一個數據集具有高基數出現，因此可緩存較小數據集的map端連接。如果兩個數據集都是高基數的，那么將面臨一個昂貴的笛卡爾積運算，執行起來會很慢，因為它不適合MapReduce的工作方式(這意味著它本身不可拆分和可并行化)。在這種情況下，我們應該重新檢查是否有任何技術(如過濾或投影)可幫助減少執行join所需的時間。

六、處理由散列分區生成的偏差

MapReduce的默認分區程序是一個散列分區程序，接受每個map輸出key的散列，并對reducer數量建模，以確定key被發送到哪個reducer。散列分區程序可以很好地用作通用分區程序，但是有些數據集可能會導致散列分區程序因一些不成比例的密鑰散列到同一個reducer而使其重載。與大多數reducer相比，這些reducer需要更長時間才能完成。此外，當檢查straggler reducer計數器時，會注意到發送給落后者的組數遠遠高于已完成的其他組。

區分高基數key與散列分區引起的偏差可以使用MapReduce reducer來識別數據傾斜類型。由性能較差的哈希分區器引入的偏差將具有更多的組(唯一密鑰)發送到這些reducer，而導致傾斜的高基數密鑰可以通過所有reducer中大致相等數量的組來證明，傾斜越多，reducer的記錄數量越多。

我們要解決的問題是reducer端連接需要很長時間才能完成，而落后的組需要比大多數reducer更長時間。使用范圍分區程序或編寫自定義分區程序，將偏移的key集中到一組reducer。此解決方案的目標是省去默認的散列分區程序，并將其替換為可以更好處理數據傾斜的內容，本文提供兩個選項可供探索：

• 使用與Hadoop捆綁在一起的sampler和TotalOrderPartitioner，將散列分區程序替換為范圍分區程序。

• 編寫自定義分區程序，將具有數據傾斜的key路由到為傾斜key保留的Reducer。

范圍分區法

范圍分區根據預定義值分配map輸出，其中每個map接收該范圍內的所有reducer，這正是TotalOrderPartitioner的工作原理。實際上，TeraSort使用TotalOrderPartitioner在所有Reducer之間均勻分布，以最大限度減少數據傾斜。TotalOrderPartitioner附帶采樣器，可對輸入數據進行采樣并將其寫入HDFS，然后在分區時由TotalOrderPartitioner使用。

自定義分區法

如果已經知道哪些Key顯示數據傾斜，并且該組Key是靜態的，則可以編寫自定義分區程序以將這些高基數key推送到一組reducer。

看完上述內容是否對您有幫助呢？如果還想對相關知識有進一步的了解或閱讀更多相關文章，請關注億速云行業資訊頻道，感謝您對億速云的支持。