當技術團隊開始將現有系統和EDH（企業數據中心）集群拼接在一起時，通常會采用以下常見的設計模式：將文件轉儲（通常為CSV格式）定期上傳到EDH中，接著進行解壓縮，轉換為最佳查詢格式，然后隱藏在HDFS中，在這里各種EDH組件都可以使用它們。

當這些文件轉儲很大或很經常出現時，這些簡單的步驟可能會顯著減慢數據擷取管道的速度。這種延遲的一部分是不可避免的；由于物理限制因素，跨網絡移動大文件是非常耗時的一件工作，并且提升其速度是非常困難的。然而，上述的其他基本數據攝取工作流程通?？梢赃M一步改進。

在這里我們向大家展示一個EDH中文件處理的簡單使用案例：在 hdfs:///user/example/zip_dir/ 中存在一個CSV文件目錄，但是該文件目錄已壓縮為原始 *.zip文件。為了使它們可用，需要將它們提取并壓縮成單個文本文件，該文件將放在 hdfs:///user/example/quoteTable_csv/中。

由于這些都是CSV文件，我們假設每個CSV文件在其第一行都有一個簡單的標題。執行此操作的一個常用方法是：在EDH的“邊緣節點”上執行一條類似于下面詳述的腳本程序 - 該“邊緣節點”是集群中的一個節點，其具有所有必需的配置文件和應用程序庫，以便與集群的其余部分進行交互。有關我們用于這些案例的邊緣節點和集群的詳細信息，請參見本文以下部分中標題為“集群詳細信息”的章節。

下圖顯示了此解決方案的基本流程，其中箭頭表示要將數據復制到位于新位置上的文件中。換句話說，塊之間的每個箭頭表示數據從左側塊復制到右側塊所需的時間。紫色箭頭表示對數據執行計算的時間，而紅色箭頭表示簡單地復制數據所需的時間。

• DataNode Disk DataNode（磁盤）
• Edge Node Disk（邊緣節點磁盤） 
• Key  （箭頭）
• Copy （復制）
• PROCESS （處理）

雖然這個解決方案是非常常見且容易實現的，但顯然存在一定的瓶頸。在我們的示例集群中，此腳本程序耗費了125秒的時間來完成包含10,000,000條記錄的zip文件。

更優的解決方案

通過利用Spark進行分發，我們可以使用相同數量的代碼更快地獲得相同的結果。通過在整個過程中將數據保存在HDFS中，我們能夠在大約36秒的時間內擷取與之前相同的數據。讓我們來看看Spark代碼，其形成了與上面顯示的bash腳本相同的結果 - 注意該代碼和本文中引用的所有代碼的更高參數化版本可以在下文中的“參考資料”章節找到。

提交到集群的程序如下圖所示：

如下圖所示，通過將這個數據擷取工作載荷從邊緣節點腳本程序移動到Spark應用程序，我們看到了顯著的速度提升 - 在示例集群上解壓縮文件所需的平均時間減少了35.7秒，這相當于速度提升超過300％。下圖顯示了在多個不同輸入上運行這兩個工作流程的結果：

對于較大的數據集而言，Spark工作流程與簡單的bash工作流程相比一般會提升超過900％的速度?，F在，我們將檢查一個更加復雜的工作流程，其中涉及解壓縮文件的處理。在此工作流程中，來自 hdfs:///user/example/zip_dir/ 的壓縮 *.csv文件的行將被解壓縮并放入Impala表quoteTable中，該表是由位于hdfs:///user/example/quoteTable/的parquet 文件提供支撐的。此外，根據數值將過濾掉其中某些行。我們先前的bash腳本程序仍然可以繼續使用，同時調用Impala將*.csv文件轉換為parquet文件：

盡管Impala執行數據轉換和過濾的速度相當快，但這種常見的使用模式仍然需要在HDFS之間復制數據。此問題如下圖所述，其中描述了這個新的工作流程：

在我們的數據集上運行上面定義的bash腳本程序138.5秒后，通過比較，我們可以修改我們的Spark作業，通過新的功能重寫下面的內容，以此實現同樣的效果：

圖中，這個程序與之前的看起來沒有任何區別 - 其中箭頭“處理”表示更密集，因為其現在包括過濾和轉換以及解壓縮操作，但是數據不會被再次寫入磁盤。另外還有一個好處，過濾掉的數據不會再被復制到磁盤中，而在我們以前的解決方案中不是這樣的。

這個Spark作業在64秒內完成，比基于bash腳本程序的解決方案速度提升了200％。對于較大的100M記錄數據集而言，我們的Spark作業速度提升超過300％。我們集群中的數據節點每個只包含2個磁盤，并且每個磁盤有足夠的內核支持2個單核執行器。通過使用功能更為強大的數據節點，對于像我們這樣的工作載荷而言，Spark對于將多線程寫入到parquet文件的支持將使其顯示比Impala更大的優勢。即使是小型集群，Spark展現出的性能優勢也是非常明顯的：

• Decompression + Conversion Runtimes （解壓縮+轉換運行時間）
• Spark runtime (s) （Spark運行時間 (s)）
• Bash runtime (s) （Bash運行時間 (s)）
• 10M records / 10 files （10M記錄/10個文件）

一旦Spark將信息加載到DataFrame中，就可以很容易地在內存中執行任何額外的轉換操作。在我們的最后一個示例中，讓我們想象一個更復雜的流程管道：我們的數據集中現在包含多個列，其中兩個采用引號括起來的字符串列可能包含我們的分隔符（‘,’），其中一個需要括起來的整數列在-100和100之間，另一個是需要平方的雙列，并且需要應用幾個簡單的過濾器。我們將使用Apache Commons CSV庫來簡單地處理更復雜輸入的解析。這個過程的Spark實現如下所示：

由于涉及到寫入更簡潔的數據類型，其最終測試的完成時間比上一個測試明顯快了很多。我們的Spark工作流程在52秒內就完成了，與傳統解決方案相比應用了少得多的代碼，傳統解決方案需要148秒才能完成。下圖顯示了上例中使用的相同數據集所需的運行時間：

如上圖所示，與使用bash和Impala的更直觀的解決方案相比，在我們的示例中數據擷取工作流程明顯速度更快，并且隨著輸入數據量的增加這種速度差異會變得更大。通過充分挖掘Spark的潛力來簡明地執行分布式計算以及以分布式方式執行定制化或第三方代碼，我們在最后一個示例中的數據擷取過程速度提升率超過600％。

現在你已經了解了其基礎知識，那么就趕快思考一下如何利用Spark加速您的ETL吧！

集群詳細信息

• 硬件： 6個EC2c3.xlarge節點
• CDH版本： 5.8.2

資源

• 腳本程序
• Spark
• 結果

歡迎撥打慧都熱線023-68661681或咨詢，我們將幫您轉接大數據專業團隊，并發送相關資料給您！