Hive SQL 遷移 Spark SQL 在網易傳媒的實踐

引言：把基於 mapreduce 的離線 hiveSQL 任務遷移到 sparkSQL，不但能大幅縮短任務運行時間，還能節省不少計算資源。最近我們也把組內 2000 左右的 hivesql 任務遷移到了 sparkSQL，這裏做個簡單的記錄和分享，本文偏重於具體條件下的方案選擇。

遷移背景

• SQL 任務運行慢

  Hive SQL 處理任務雖然較爲穩定，但是其時效性已經達瓶頸，無法再進一步提升，同一個 SQL，Hive 比 Spark 執行的時間更長。

• Spark SQL 的發展遠超 HSQL

  隨着 Spark 以及其社區的不斷髮展，Spark SQL 本身技術也在不斷成熟，Spark 在技術架構和性能上都展示出 Hive 無法比擬的優勢。

遷移方案設計

在遷移過程中，首先需要確認如何解決執行引擎 sql 語法不兼容以及執行結果不一致的問題。常用的做法是在執行引擎層面進行兼容，用戶無感知，這是較爲理想的方法。但對於我們來說，這是一個不可選項，組內基本工作在頂層業務層面，沒有對底層 spark 引擎特別熟的同事，數科同事也不會對我們的祖傳代碼負責，所以我們只能通過修改業務 sql 代碼解決兼容性問題。

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

• 如何將任務執行引擎切換爲 spark

• 如何測試任務，並驗證數據一致性

• 如何自動化處理

這些問題明顯是與任務的具體運行形式相關的，所以有必要先介紹下現有的 hiveSQL 的運行方式。現有的 hiveSQL 任務主要以 shell 腳本的方式提交，腳本代碼結構如下所示：

#!/bin/bash
#結構1
#根據腳本參數、環境變量等計算時間參數
dayStr='xxx'
#結構2
#根據時間參數拼接hiveSql字符串
sqlStr="insert overwrite t1 partition(day='${dayStr}') select XXX from t2 where day='${dayStr}';"
#結構3
#運行hiveSQL
hive -e "${sqlStr}"
#結構4
#可選，將結果導入MySQL、impala等存儲系統

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

如何將任務執行引擎切換爲 spark，是分歧最大的地方。

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

選擇 kyuubi 的決定是非常正確的。首先，因爲我們希望使用抽象的 sparkSQL 運行服務，把 SQL 扔給它就完了，業務代碼和 spark 部署、任務運行環境等細節完全解耦，kyuubi 在這方面非常合適。其次 kyuubi 本身的功能要遠超 Spark Thrift Server，主要集中在可以隔離不同的 SQL 任務、支持高可用、自動優化任務等方面，能讓用戶偷懶的組件就是好組件。最後的原因是在人員合作方面，使用 kyuubi 等於找到了一個穩定的背鍋俠，出了問題直接讓數科 spark 團隊處理就完了，溝通非常直接，如果使用的是 spark 客戶端或中臺的 SQL 運行功能，中間可能隔着好幾個人。

反映到具體代碼細節上，我們首先編寫了一個名爲 sparkUtils.sh 的腳本，裏面定義了所有與 sparkSQL 相關的實用函數，比如 runSql、setResource 等，這樣其他腳本只需要 source 一下這個腳本，然後將 hive -e 改爲 runSql 就可以將任務執行方式改爲 spark。

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

爲確定任務是否可上線，必須計算出數據的一致程度。這裏的採用的方法比較簡單，先計算測試表分區和線上表分區相同的數據條數，然後算出一致度，比如線上分區數據是 2 條，測試分區數據是 8 條，相同數據是 1 條，那麼一致度是（1*2）/（2+8）=0.2。爲方便確定哪些列的數據不一致，我們計算了列 hash。

切換執行引擎、測試、上線的方案都確定後，最後一個問題便是如何實現自動化處理，畢竟有 2000 個任務，全靠人工處理工作量太大。但由於 shell 腳本代碼並不是結構化的數據，全自動處理也不現實，線上數據出問題的鍋可背不起。

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

除了 3 個代碼分支，這裏多了一個測試機，用於自動化測試並計算數據一致性。各步驟具體介紹如下：

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

• 執行方式切換爲 spark：自動化處理，人工審覈後提交；

• 生成測試代碼：通過正則匹配 insert 語句，替換爲臨時表，並在之前插入創建臨時表語句，註釋其他影響線上數據的操作。自動化處理，人工審覈提交；

• 拉取代碼、測試、計算一致程度：完全自動化，測試後本地生成一個 csv 的測試報告；

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

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

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

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

遷移成果

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

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

任務遷移前後的運行情況如下圖所示，橫軸代表任務運行時間，縱軸代表任務累計 CPU 使用（CPU 核數 * 分鐘）。這裏共選取了 600 多個任務，藍色的點代表遷移前 MapReduce 的運行情況，紅色點代表遷移後 sparkSQL 的運行情況。根據統計，遷移後平均運行時間節省 70%，CPU 累計使用平均節省 35%。

總的來看，收益超過預期，特別是在運行時間方面。

總結

回顧整個方案的設計過程，實際上沒有太多選擇的餘地，在沒法在 spark 引擎層做兼容的前提，和以腳本提交任務的現狀下，只能選擇基於 git 版本管理的自動化遷移流程。

方案能這麼順利實施，主要因爲任務代碼是以腳本的形式存在，這樣我們可以很方便的用各種程序處理腳本源代碼，避免了大量重複性的工作，特別是用 git 進行版本管理，如果我們的任務都寫在了公司中臺上，那麼遷移工作量會大很多。

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

作者簡介

易同學，網易傳媒大數據開發工程師，2020 年入職網易，負責數據接入數倉、離線 spark 混部、資源治理等方面。

Data 跳動 專注於圖解各種技術棧，Kafka、Hadoop、Hive、Spark、Flink 等

本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/079KkPEr0YbzduByPvLRYw