Airbnb開源ReAir工具，提供PB級數據倉庫的遷移和備份
Airbnb 大數據平臺架構成為 Airbnb 公司提升產品決策的關鍵部分。其 Hive 數據倉庫從 2013 年中旬的 350 TB 暴增到 11 PB （2015 年末統計的數據）。隨著公司的成長，數據倉庫的可靠性需求日益劇增。我們尋求遷移數據倉庫，但現有的遷移工具要么在大數據倉庫時有問題，要么就是有很明顯的操作負荷，所以 Airbnb 開發了 ReAir 解決這種狀況。這篇文章將詳細介紹 ReAir 是如何工作的以及它是怎樣輕松的實現 PB 級數據倉庫的備份。
背景
最開始 Airbnb 所有數據存放在單個 HDFS／HIve 數據倉庫。單個命名空間簡單、易管理，然而產品復雜后之后即席查詢就影響其可靠性。因此，我們把數據倉庫分成了兩個：一個數據倉庫是支持關鍵產品的任務；另一個是專為即席查詢用的。分離這么大的數據倉庫將面臨兩個問題：我們如何容易的遷移龐大的數據倉庫？；分離完之后，我們又該怎么保持數據的同步？為了解決這些問題，Airbnb 開發 ReAir 項目并開源給社區。
ReAir 對基于 Hive 元數據的數據倉庫進行備份非常有效，可以實現 PB 級別的集群擴展。使用 ReAir 極其簡單，你只要連接 Hive 的 metastore Thrift 服務，通過 MapReduce 進行數據復制。ReAir 可以兼容 Hive 和 Hadoop 的各種版本，并支持單機模式操作——只有 Hadoop 和 Hive 是必須要有，MySQL DB 是用來進行增量復制。由于 ReAir 同時處理數據和元數據，所以，只要復制任務一完成就可以進行表和分區的查詢。
在沒有 ReAir 的時候，遷移 Hive 數據倉庫最典型的用法是啟動一個 DistCp，并且通過數據庫操作手動管理元數據。這種方法既費力又容易出錯，甚至會引起臟數據導致數據不一致，有時復制不斷變化的數據倉庫目錄時也會出現問題。另外其它的遷移方法要求指定 Hive 版本，當我們需要從舊版本的 Hive 數據倉庫遷移就變得異常困難。
ReAir 工具包含兩種遷移工具：批量遷移和增量式遷移。批量式遷移工具允許你一次復制指定的一系列表，它適合對一個集群的遷移。相對應的，增量式遷移工具可以跟蹤數據倉庫的變化，并復制生成的對象或者修改的對象。增量式遷移工具適合保持集群間數據的同步，它可以實現秒級數據同步。兩種方式更詳細的對比請見下面部分。
批量遷移
批量式遷移一般用來備份整個數據倉庫。復制的速度和吞吐量取決于 reducer 的數量和吞吐量，在 Airbnb 公司，使用 500 個 reducer 復制 2.2 PB 數據僅僅用了 24 小時。
啟動批量遷移的過程非常簡單：用戶允許 shell 命令啟動一系列 MR 任務。其執行協議是從源數據倉庫復制 Hive 表和分區等實體到目的數據倉庫。批量式復制過程是占有帶寬的，它會探測源數據倉庫和目的數據倉庫，并只復制兩者之間的不同文件。同時元數據也僅僅更新變化過的。這些策略可以保證 ReAir 工作的高效率。
批量式遷移也有一些挑戰。最明顯的一個是，產品數據倉庫中實體（Hive 表和分區）的大小不同，但是遷移的延遲并不能依賴于最大的一個實體。例如，一般的 Hive 表不到 100 個分區，然后最大的表有超過 10 萬個分區。為了保證正常的延遲，需要并行運行遷移工作。
為了解決負載均衡的問題，批量式遷移運行一系列 MR 任務。批量遷移中兩個最昂貴的操作是文件復制和元數據更新，所以這些步驟都是在 shuffle 階段分布式進行。每個任務都會打印運行日志數據存儲在 HDFS，通過日志數據能清晰的看到任務的完成情況。
第一個 MR 任務從 HDFS 讀取實體標識符并 shuffle 后均勻的發到 reducer。reducer 先驗證各實體，再對要復制的 HDFS 目錄和實體進行映射；第二個 MR 任務掃描第一個 MR 產生的文件目錄，并在對應的目錄下創建文件。對文件名稱進行 hash shuffle 后發到 reducer，一旦 shuffle 之后 reducer 執行復制操作；第三個 MR 任務處理 Hive 元數據的提交邏輯。
三階段 MR 任務計劃的擴展性和負載均衡都很好：復制 1 百萬實體的 2.2 PB 數據消耗大改 24 小時；同步 20 TB 數據量的更新僅花了一個小時。階段 1 和階段 3 的瓶頸在于 Hive 元數據 MySQL 數據庫；而階段 2 的瓶頸在于網絡帶寬。
對于我們的遷移，需要開發定制化的文件復制 MR 來處理 HDFS 上的數據。然而，對比 DistCp 這樣通用化的工具，在測試 ReAir 過程中也發現了一些問題：
- 在復制百萬個文件或者整個數據倉庫時，MR 任務初始化比較慢；
- 錯誤率較高，并且需要定制錯誤處理；
- 卻少易用的日志分析功能。
為了解決這些問題，我們開發了兩個 MR 任務來處理通用 HDFS 數據復制。第一個任務采用一個啟發式的、文件夾遍歷的多線程進行一系列分隔。一旦有足夠的文件夾，mapper 遍歷這些文件夾生成文件列表來復制。文件名經過 shuffle 發送到 reducer 來決定是否有必要復制。第二個 MR 任務讀取文件列表來復制，并通過一個 shuffle 進行分布式復制工作。
增量式遷移
在兩個集群的情況下，我們需要在兩個集群之間共享數據。例如，日志數據需要每天在生產集群聚會，但同時即席查詢的用戶也需要這些數據。批量式復制任務對于這種需求顯得太重，這時需要兩個數據倉庫間可以按小時進行更新。即席查詢集群的用戶需要盡快同步生產集群的數據，因此有必要找到一個盡可能快的方法來更新新的內容。雖然有一些開源項目能解決這個問題，但由于 Hive 版本依賴等問題并不是很理想，后來我們開發增量式復制工具來保證即席查詢集群和生產集群的數據同步。
增量式復制工具設計到實體（Hive 表和分區）的記錄變化，一旦發生變化盡快復制變化量。為了記錄源集群上的變化，我們使用了 Hive 的鉤子機制（hook 函數）把查詢成功的實體寫入 MySQL 數據庫。采用這種方法，我們可以跟蹤生產集群的所有變化。HDFS 上的變化或者元數據的更新都能觸發復制。
一旦有了數據庫里的變化的記錄，我們需要一種方法來復制這些變化到即席查詢集群。這個執行機制是通過一個 Java 服務實現，讀取變化的日志中實體，并轉化成一系列的行為集合。比如，在源集群成功創建一個表，會在目標集群上翻譯成“復制表”的行為。Java 進程將調用元數據，并啟動 MR 任務執行。
由于源集群上的變化是序列化到日志里，它將在目的集群中以相同的順序進行執行。但是，實際情況是，單獨的復制一個表或者分區花費幾秒鐘或者幾分鐘實在太慢，后面改成多線程并行復制。為了處理并發，所有的行為基于并發限制形成 DAG。通常限制都是對于同一個表，比如，在復制分區之前要建立好表。通過并發執行，復制的延遲降低到最小。
用增量式復制可以實現快速、可靠的復制。兩個數據倉庫之間的同步復制可以實現容災恢復——如果一個集群宕機，另外一個集群還可以正常提供服務。對比批量式復制，增量式復制對數據倉庫體量巨大但改變的數據量比較小的情況更有效率。當前在 Airbnb，每天的數據量增長不到 2 TB，所以增量式復制比較有意義。
使用批量式遷移和增量式遷移，可以快速的遷移兩個集群。我們希望這些工具對社區也一樣有用。
來自: InfoQ