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

當(dāng)這些文件轉(zhuǎn)儲很大或很經(jīng)常出現(xiàn)時，這些簡單的步驟可能會顯著減慢數(shù)據(jù)擷取管道的速度。這種延遲的一部分是不可避免的；由于物理限制因素，跨網(wǎng)絡(luò)移動大文件是非常耗時的一件工作，并且提升其速度是非常困難的。然而，上述的其他基本數(shù)據(jù)攝取工作流程通常可以進(jìn)一步改進(jìn)。

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

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

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

• DataNode Disk DataNode（磁盤）
• Edge Node Disk（邊緣節(jié)點(diǎn)磁盤） 
• Key  （箭頭）
• Copy （復(fù)制）
• PROCESS （處理）

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

更優(yōu)的解決方案

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

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

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

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

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

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

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

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

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

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

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

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

現(xiàn)在你已經(jīng)了解了其基礎(chǔ)知識，那么就趕快思考一下如何利用Spark加速您的ETL吧！

集群詳細(xì)信息

• 硬件： 6個EC2c3.xlarge節(jié)點(diǎn)
• CDH版本： 5.8.2

資源

• 腳本程序
• Spark
• 結(jié)果

歡迎撥打慧都熱線023-68661681或咨詢，我們將幫您轉(zhuǎn)接大數(shù)據(jù)專業(yè)團(tuán)隊，并發(fā)送相關(guān)資料給您！