Spark 實(shí)踐 | Hive SQL 遷移 Spark SQL 在網(wǎng)易傳媒的實(shí)踐

• SQL任務(wù)運(yùn)行慢

  Hive SQL處理任務(wù)雖然較為穩(wěn)定，但是其時(shí)效性已經(jīng)達(dá)瓶頸，無法再進(jìn)一步提升，同一個SQL，Hive比Spark執(zhí)行的時(shí)間更長。

• Spark SQL的發(fā)展遠(yuǎn)超HSQL

  隨著 Spark 以及其社區(qū)的不斷發(fā)展，Spark SQL 本身技術(shù)也在不斷成熟，Spark在技術(shù)架構(gòu)和性能上都展示出Hive無法比擬的優(yōu)勢。

在遷移過程中，首先需要確認(rèn)如何解決執(zhí)行引擎sql語法不兼容以及執(zhí)行結(jié)果不一致的問題。常用的做法是在執(zhí)行引擎層面進(jìn)行兼容，用戶無感知，這是較為理想的方法。但對于我們來說，這是一個不可選項(xiàng)，組內(nèi)基本工作在頂層業(yè)務(wù)層面，沒有對底層spark引擎特別熟的同事，數(shù)科同事也不會對我們的祖?zhèn)鞔a負(fù)責(zé)，所以我們只能通過修改業(yè)務(wù)sql代碼解決兼容性問題。

確定了修改SQL代碼的兼容方式后，接下來便需要在此基礎(chǔ)上確定以下問題的解決方法：

• 如何將任務(wù)執(zhí)行引擎切換為spark

• 如何測試任務(wù)，并驗(yàn)證數(shù)據(jù)一致性

• 如何自動化處理

這些問題明顯是與任務(wù)的具體運(yùn)行形式相關(guān)的，所以有必要先介紹下現(xiàn)有的hiveSQL的運(yùn)行方式。現(xiàn)有的hiveSQL任務(wù)主要以shell腳本的方式提交，腳本代碼結(jié)構(gòu)如下所示：

#!/bin/bash#結(jié)構(gòu)1#根據(jù)腳本參數(shù)、環(huán)境變量等計(jì)算時(shí)間參數(shù)dayStr='xxx'#結(jié)構(gòu)2#根據(jù)時(shí)間參數(shù)拼接hiveSql字符串sqlStr="insert?overwrite?t1?partition(day='${dayStr}')?select?XXX?from?t2?where?day='${dayStr}';"#結(jié)構(gòu)3#運(yùn)行hiveSQLhive?-e?"${sqlStr}"#結(jié)構(gòu)4#可選，將結(jié)果導(dǎo)入MySQL、impala等存儲系統(tǒng)

腳本的好處在于可以通過git管理代碼的變更、通過IDE搜索代碼引用，壞處在于沒有和公司的中臺進(jìn)行整合，很多功能用不上。

如何將任務(wù)執(zhí)行引擎切換為spark，是分歧最大的地方。

第一種選擇是直接將SQL任務(wù)遷移到公司中臺上，這樣可以直接在平臺選擇執(zhí)行引擎，讓平臺幫我們屏蔽掉任務(wù)的運(yùn)行細(xì)節(jié)。這個方法也不是不行，但工作量和難度都很大，很快就被排除了。為控制工作量，保持任務(wù)仍以腳本的形式運(yùn)行是必須的，所以SQL任務(wù)的提交方式也只能通過shell命令。如何在shell中提交sparkSQL也有2個選擇，第一種是直接在我們的az調(diào)度機(jī)上部署spark客戶端，然后通過本地的spark客戶端運(yùn)行任務(wù)，另一種則是通過beeline將SQL提交到類似于hiveserver的服務(wù)上。考慮之后，我們選擇了第二種，具體來說，是使用了網(wǎng)易有數(shù)姚老師、尤老師等大佬的作品kyuubi。

選擇kyuubi的決定是非常正確的。首先，因?yàn)槲覀兿Ｍ褂贸橄蟮膕parkSQL運(yùn)行服務(wù)，把SQL扔給它就完了，業(yè)務(wù)代碼和spark部署、任務(wù)運(yùn)行環(huán)境等細(xì)節(jié)完全解耦，kyuubi在這方面非常合適。其次kyuubi本身的功能要遠(yuǎn)超Spark Thrift Server，主要集中在可以隔離不同的SQL任務(wù)、支持高可用、自動優(yōu)化任務(wù)等方面，能讓用戶偷懶的組件就是好組件。最后的原因是在人員合作方面，使用kyuubi等于找到了一個穩(wěn)定的背鍋俠，出了問題直接讓數(shù)科spark團(tuán)隊(duì)處理就完了，溝通非常直接，如果使用的是spark客戶端或中臺的SQL運(yùn)行功能，中間可能隔著好幾個人。

反映到具體代碼細(xì)節(jié)上，我們首先編寫了一個名為sparkUtils.sh的腳本，里面定義了所有與sparkSQL相關(guān)的實(shí)用函數(shù)，比如runSql、setResource等，這樣其他腳本只需要source一下這個腳本，然后將hive -e改為runSql就可以將任務(wù)執(zhí)行方式改為spark。

在解決如何運(yùn)行任務(wù)后，接下來便是解決如何測試任務(wù)的問題。由于沒有隔離的測試環(huán)境，所以在測試任務(wù)時(shí)，必須將任務(wù)代碼無害化，包括替換insert的表為臨時(shí)表、注釋任何可能影響線上數(shù)據(jù)的操作。這樣一來，同一個腳本，我們至少需要有3個版本：線上的hiveSQL版、未上線的sparkSQL版、無害化的使用sparkSQL運(yùn)行的測試版，而且我們需要在他們之間同步代碼更改，比如測試途中，線上版有更改，那么我們需要將更改合并到sparksql版和測試版，如果測試中發(fā)現(xiàn)不兼容問題，需要先在sparksql版上修改，然后合并到測試版，最后上線操作則是使用sparksql版覆蓋線上版。整個過程基本以git版本管理為核心。

為確定任務(wù)是否可上線，必須計(jì)算出數(shù)據(jù)的一致程度。這里的采用的方法比較簡單，先計(jì)算測試表分區(qū)和線上表分區(qū)相同的數(shù)據(jù)條數(shù)，然后算出一致度，比如線上分區(qū)數(shù)據(jù)是2條，測試分區(qū)數(shù)據(jù)是8條，相同數(shù)據(jù)是1條，那么一致度是（1*2）/（2+8）=0.2。為方便確定哪些列的數(shù)據(jù)不一致，我們計(jì)算了列hash。

切換執(zhí)行引擎、測試、上線的方案都確定后，最后一個問題便是如何實(shí)現(xiàn)自動化處理，畢竟有2000個任務(wù)，全靠人工處理工作量太大。但由于shell腳本代碼并不是結(jié)構(gòu)化的數(shù)據(jù)，全自動處理也不現(xiàn)實(shí)，線上數(shù)據(jù)出問題的鍋可背不起。

所以我們制定的策略是半自動化，人工審核代碼修改，其他部分盡可能自動化，特別是測試和計(jì)算數(shù)據(jù)一致性部分。總的工作流如下圖所示：

除了3個代碼分支，這里多了一個測試機(jī)，用于自動化測試并計(jì)算數(shù)據(jù)一致性。各步驟具體介紹如下：

• 檢測未測試代碼：通過公司中臺api獲取有調(diào)度的腳本，拷貝到測試庫，完全自動化；

• 執(zhí)行方式切換為spark：自動化處理，人工審核后提交；

• 生成測試代碼：通過正則匹配insert語句，替換為臨時(shí)表，并在之前插入創(chuàng)建臨時(shí)表語句，注釋其他影響線上數(shù)據(jù)的操作。自動化處理，人工審核提交；

• 拉取代碼、測試、計(jì)算一致程度：完全自動化，測試后本地生成一個csv的測試報(bào)告；

• 判斷是否可上線：人工審核；

• 修改不兼容語法：基本靠人工修改，一些簡單的參數(shù)替換可自動化；

• 合并修改：人工處理，不過基本可以自動合并；

• 刪除測試代碼，代碼上線：自動處理，人工審核提交。

如下圖所示，在2個月的時(shí)間內(nèi)，我們spark程序所使用的資源從10%左右上升到了80%左右。

CPU累計(jì)使用（CPU核數(shù)*分鐘）的變化曲線如下所示。由于遷移過程中也下線了不少任務(wù)，所以CPU總量的下降不光是遷移的功勞。

任務(wù)遷移前后的運(yùn)行情況如下圖所示，橫軸代表任務(wù)運(yùn)行時(shí)間，縱軸代表任務(wù)累計(jì)CPU使用（CPU核數(shù)*分鐘）。這里共選取了600多個任務(wù)，藍(lán)色的點(diǎn)代表遷移前MapReduce的運(yùn)行情況，紅色點(diǎn)代表遷移后sparkSQL的運(yùn)行情況。根據(jù)統(tǒng)計(jì)，遷移后平均運(yùn)行時(shí)間節(jié)省70%，CPU累計(jì)使用平均節(jié)省35%。

總的來看，收益超過預(yù)期，特別是在運(yùn)行時(shí)間方面。

回顧整個方案的設(shè)計(jì)過程，實(shí)際上沒有太多選擇的余地，在沒法在spark引擎層做兼容的前提，和以腳本提交任務(wù)的現(xiàn)狀下，只能選擇基于git版本管理的自動化遷移流程。

方案能這么順利實(shí)施，主要因?yàn)槿蝿?wù)代碼是以腳本的形式存在，這樣我們可以很方便的用各種程序處理腳本源代碼，避免了大量重復(fù)性的工作，特別是用git進(jìn)行版本管理，如果我們的任務(wù)都寫在了公司中臺上，那么遷移工作量會大很多。

在整個遷移過程，除了前期踩坑階段，期間線上基本沒出什么問題，十分平滑的將2000左右的任務(wù)遷移到了sparkSql，而且也沒耗費(fèi)過多人力，這說明整個遷移方案的設(shè)計(jì)和實(shí)施是比較成功的。

分享，點(diǎn)贊，在看，安排一下？