Impala 3.4 在網(wǎng)易的最新實踐
導(dǎo)讀：Impala是Cloudera公司主導(dǎo)開發(fā)的交互式查詢系統(tǒng)，它提供SQL語義和計算能力，但是本身并不存儲數(shù)據(jù)。本次分享會聚焦于Impala在網(wǎng)易內(nèi)部的一些新實踐，以及基于Impala 3.4版本所做的優(yōu)化和改進。
主要會圍繞以下四點展開：
Impala定位及使用
Impala對接Iceberg
Impala管理系統(tǒng)
Impala未來規(guī)劃
1. 什么是Impala？
Cloudera開源貢獻至Apache的OLAP引擎
提供高并發(fā)和低延遲的交互式SQL查詢
可以查詢HDFS/HBase/Kudu中的數(shù)據(jù)
2. Impala優(yōu)勢
去中心化的MPP架構(gòu)
完全兼容Hive元數(shù)據(jù)格式
Apache頂級項目，社區(qū)活躍度高
支持多種數(shù)據(jù)格式，例如Parquet、Orc、Avro等
高效的查詢性能，支持codegen、llvm、runtime filter等
3. Impala架構(gòu)簡介
下面是Impala的一個架構(gòu)，如圖所示：
Impalad節(jié)點分為兩種角色：
Coordinator(協(xié)調(diào)者)：響應(yīng)客戶端請求，SQL解析
Executor(執(zhí)行者)：數(shù)據(jù)掃描、聚合運算
Impalad內(nèi)部分為兩個模塊：
Frontend：執(zhí)行SQL解析，Java代碼編寫
Backend：進行聚合運算，C++代碼編寫
Impala主要分為三種服務(wù)，每種服務(wù)對應(yīng)一個單獨的進程。
第一種服務(wù)：從最左邊綠色的部分可以看起，它是一個Impalad 進程，主要有兩種角色，第一種角色：Coordinator（協(xié)調(diào)者），它主要的功能是響應(yīng)JDBC請求，對發(fā)過來的SQL進行執(zhí)行計劃解析，將SQL生成的執(zhí)行計劃發(fā)給各個Executor（執(zhí)行者），這些Executor會進行數(shù)據(jù)掃描和聚合運算等操作。每個Impalad節(jié)點，既可以充當(dāng)Coordinator，也可以充當(dāng)Executor，也可以二者兼之。在Impalad內(nèi)部也分為兩個模塊，其中FE模塊由Java代碼編寫，主要執(zhí)行SQL的解析操作，使用Java可以更好地兼容Hadoop生態(tài)圈；BE模塊由C++代碼編寫，主要負責(zé)實際的數(shù)據(jù)掃描、聚合運算，使用C++可以進行更好地性能優(yōu)化。
第二種服務(wù)：catalogd，它是一個元數(shù)據(jù)服務(wù)，單獨的一個進程，主要就是將hive的metastore所存儲的元數(shù)據(jù)緩存到自己的內(nèi)存當(dāng)中。
第三種服務(wù)：statestored，它是一個發(fā)布訂閱服務(wù)，主要的作用包括節(jié)點之間狀態(tài)信息的同步、元數(shù)據(jù)信息的同步、資源隊列的信息同步等等。
這里以元數(shù)據(jù)服務(wù)加載為例：首先catalogd將元數(shù)據(jù)緩存到自己的內(nèi)存中，同時它會將元數(shù)據(jù)信息發(fā)布到statestored上面，然后Impalad就可以去訂閱相應(yīng)的topic，將這部分信息拉到自己的本地進行緩存。這整體就是一個發(fā)布訂閱的過程。
需要注意的是，在一個Impala集群中，statestored和catalogd只有一個，Impalad節(jié)點有若干個。由于coordinator需要進行執(zhí)行計劃解析，因此需要緩存元數(shù)據(jù)在自己的內(nèi)存中。實際線上部署的時候，我們一般會將coordinator和executor分開，同時集群中只有配置少量幾個coordinator節(jié)點，大部分都是executor節(jié)點。
4. Impala 3.x新特性
3.0到3.4的特性羅列：
支持在相同的查詢塊中存在多個distinct算子
支持優(yōu)雅的（不影響正在執(zhí)行的查詢）關(guān)閉impala進程
支持ORC文件格式
支持DATE數(shù)據(jù)類型及其操作函數(shù)
支持將遠端的HDFS/S3等文件緩存到Impalad節(jié)點上
CBO增強
支持將Profile信息導(dǎo)出成JSON格式，方便解析
...
5. Impala內(nèi)部特性增強
Impala 在網(wǎng)易內(nèi)部進行的一些開發(fā)，以下做了一個簡單的列舉：
支持Impala對接Iceberg
元數(shù)據(jù)同步功能
Impala on Alluxio功能
Impala管理系統(tǒng)
集群節(jié)點分組功能
6. Impala在網(wǎng)易易數(shù)
Impala在網(wǎng)易這邊的定位是作為一個交互式查詢系統(tǒng)，對下可以查詢存儲在HDFS、Kudu、HBase等等這些存儲系統(tǒng)中的數(shù)據(jù)，向上它可以對接一些我們開發(fā)的BI工具、或者是一些業(yè)務(wù)自己的系統(tǒng)等。
上圖所示是我們內(nèi)部的一個使用場景：自助分析，主要提供給分析師或者數(shù)據(jù)開發(fā)人員使用，直接在頁面上提交SQL查詢，通過SQL來獲取自己想要的數(shù)據(jù)，這個與Cloudera的HUE類似。
另外一個使用場景就是：網(wǎng)易有數(shù)，作為一款BI工具，它提供了一個圖形化界面，用戶只需要在這個頁面上進行相應(yīng)的一些拖拽操作，系統(tǒng)會自動地生成相應(yīng)的SQL，發(fā)到Impala，根據(jù)Impala的查詢結(jié)果，以圖形化的界面形式展示給用戶。用戶不需要關(guān)注具體的SQL編寫，只需要在頁面上進行一些簡單的控件拖拽，就可以獲取相應(yīng)的圖標展示，非常方便。
1. 什么是Iceberg？
Apache Iceberg is an open table format for huge analytic datasets.
從官網(wǎng)上來看可以知道它是一個表格式，為了一個海量的數(shù)據(jù)分析所誕生的，這也就意味著它并不是一個單獨的服務(wù)，它只是提供了一系列的API，我們需要去操作這些API。
2. 為什么要引入Iceberg？
以下是我們在內(nèi)部業(yè)務(wù)中遇到的痛點，這里簡單給大家歸納以下：
① 百TB級的離線任務(wù)延遲導(dǎo)致報表產(chǎn)出時間不穩(wěn)
凌晨NameNode壓力很大，請求延遲不穩(wěn)定
任務(wù)ETL效率相對低效，一次ETL需要時間2個小時
一旦遇到磁盤壞掉或者機器宕機，Spark任務(wù)重試一次就會導(dǎo)致2小時延遲
② 基于Lambda架構(gòu)的實時數(shù)據(jù)倉庫存在較多問題
Kafka無法支持海量數(shù)據(jù)存儲，無法支持高效的OLAP查詢
Lambda架構(gòu)維護成本很高
上圖目前我們網(wǎng)易內(nèi)部的一個數(shù)倉架構(gòu)圖，可以看到，維護實時和離線兩條線路，本身涉及到的組建也比較多。關(guān)于這塊相關(guān)的情況，之前也有同事做過Iceberg相關(guān)的分享，這里就不再展開了。
3. Impala支持Iceberg功能
通過Impala創(chuàng)建Iceberg表
通過Impala查詢Iceberg表
支持INSERT INTO非分區(qū)的Iceberg表（Parquet格式）
支持部分ALTER操作，例如ADD COLUMNS/RENAME TABLE等
支持DESCRIBE HISTORY，查看表的歷史snapshots
4. Impala創(chuàng)建Iceberg表
Catalog類型：
HiveCatalog/HadoopCatalog/HadoopTables
數(shù)據(jù)文件格式：Parquet/ORC格式
分區(qū)類型：
IDENTITY/YEAR/MONTH/DAY/BUCKET/TRUNCATE
Iceberg本身提供了好幾種分區(qū)類型，它與傳統(tǒng)的Hive不太一樣。舉一個簡單的例子：
這是Impala創(chuàng)建Iceberg表的一個SQL，首先我們知道Hive的分區(qū)列只是一個邏輯上的概念，是HDFS上的一個目錄層級。但是Iceberg表的分區(qū)列的數(shù)據(jù)，在底層的數(shù)據(jù)文件中也是存在的，所以在建表的時候，分區(qū)列也必須位于表名后面。第二點是新增了一個關(guān)鍵字SPC，如果要創(chuàng)建Iceberg分區(qū)表的話，必須要指定關(guān)鍵字SPEC。第三點就是我們在定義分區(qū)的時候，列名后面跟的是分區(qū)類型，而Hive后面跟的是列的類型。最后一點需要注意的是，需要在TBLPROPERTIES中指定各個屬性。因為Iceberg在Impala中是一個表格式，有不同的數(shù)據(jù)文件，所以目前采取的方式是在表屬性中對這些屬性進行相應(yīng)的定義。總結(jié)一下，有以下幾點：
分區(qū)列數(shù)據(jù)也存在于數(shù)據(jù)文件中
新關(guān)鍵字標識SPEC
定義分區(qū)時，列名后面跟分區(qū)類型
需要在表屬性中定義文件格式、catalog類型
5. Impala查詢Iceberg表
接下來看一下Impala查詢Iceberg表簡單流程：
① SQL請求到Impalad節(jié)點，F(xiàn)E模塊進行解析；
② FE將謂詞條件下推至Iceberg，其實就是調(diào)用了Iceberg的api；
③ Iceberg根據(jù)元數(shù)據(jù)信息和謂詞條件進行篩選
④ 返回待掃描文件列表至FE；
⑤ 由FE生成執(zhí)行計劃；
⑥ 各個Impalad執(zhí)行實際的掃描，返回結(jié)果；
⑦ FE將結(jié)果返回至客戶端；
總結(jié)下來就是：將Iceberg表當(dāng)成一種特殊的HDFS表，使用謂詞通過Iceberg API過濾待掃描文件列表，后續(xù)操作步驟按照HDFS表處理。這樣設(shè)計的好處，就是可以復(fù)用大量的Impala代碼（Impala的SCAN都是在BE模塊，用C++實現(xiàn)的）。
6. 其他SQL支持
① INSERT INTO非分區(qū)表（Parquet）
INSERT INTO xxx SELECT * FROM xxx
INSERT INTO xxx VALUES(...)
CREATE TABLE xxx AS SELECT * FROM xxx
② DESCRIBE HISTORY查看snapshots
DESCRIBE HISTOR Y xxx
7. 2.12.0 parquet vs Iceberg
我們將基本的建表和查詢功能backup到了2.12.0版本，與parquet進行了對比，測試環(huán)境為10臺384G、48核的物理機，數(shù)據(jù)集為TPCDS-1000，測試結(jié)果如下：
目前我們只是做了一個簡單的對比，后續(xù)會進行更深入的分析和優(yōu)化。
8. 社區(qū)相關(guān)動態(tài)
社區(qū)EPIC地址：
https://issues.apache.org/jira/browse/IMPALA-10149
目前開發(fā)基于社區(qū)4.0版本，低版本需要手動將patch合并
目前還有很多功能待完成，大家感興趣的話，歡迎大家一起參與開發(fā)。
1. Impala管理系統(tǒng)增強
管理系統(tǒng)是網(wǎng)易基于Impala開發(fā)的一個服務(wù)，是一個單獨的進程， 用于將線上集群的所有查詢信息持久化到數(shù)據(jù)庫，我們這里主要介紹以下，基于管理系統(tǒng)所新增的三個模塊。
Profile解析模塊
Compute stats模塊
資源隊列配置模塊
2. Profile解析模塊
從上面這個截圖可以看到，profile其實是非常詳細的，它包含了整個查詢的所有相關(guān)信息。
所有信息都存在于一個Profile里面， 每次進行SQL分析都需要去Profile 里面查詢，不利于線上運維
Impala 3.4版本提供了json格式的 profile文件下載
基于這兩點考量，網(wǎng)易開發(fā)了一個的解析模塊，它主要包括下面兩個功能：
解析Profile中的相關(guān)信息
解析SQL的Summary信息
接下來就分別看看這兩點：
① Node解析
Node分為ScanNode、JoinNode、SortNode。上圖是典型的HTFS_SCAN_NODE，它包含了很多的指標。從截圖中看到，像讀取的字節(jié)數(shù)，本地讀取的字節(jié)數(shù)，讀取的函數(shù)等等，這塊都是有展示的。我們將這些指標都提取出來，然后解析成一條一條的記錄存儲到Mysql表當(dāng)中。
通過對這些字段進行一些過濾、排序和篩選，例如：
Bytes Read最多的Top N的SQL
nOpen HDFS File耗時最長的Top N SQL
......
② Summary解析
上面是summary信息，它包含了SQL執(zhí)行每個階段，我們將這些階段也都進行相應(yīng)的解析提取，然后轉(zhuǎn)化成一條一條的記錄，然后存儲在MySQL表中。
和上面一樣，我們也可以對表進行相應(yīng)的一些處理，獲取如下的一些信息：
SCAN HDFS最慢的Top N SQL
HASH JOIN最慢的Top N SQL
......
3. Compute Stats模塊
統(tǒng)計信息計算模塊，Impala之前對于統(tǒng)計信息的依賴是非常重的，如果說這個表沒有統(tǒng)計信息的話，它的執(zhí)行計劃有時候會非常的差，尤其是我們Join的方式。我們所說的統(tǒng)計信息通常會包括像每個列的最大值，最小值，還有它的distinct值等，這些我們都可以稱之為統(tǒng)計信息。基于這個原因，網(wǎng)易目前開發(fā)了這樣一個模塊，它主要包括這三個功能：
指定時間、指定表，進行compute stats
可以進行全量/增量進行compute stats
compute stats的結(jié)果進行展示
下面就是一些相關(guān)的配置和頁面展示情況：
4. Impala 3.4優(yōu)化新參數(shù)
下面兩個新的參數(shù)3.4版本所引入的，也可以理解是對于CBO的增強。
① BROADCAST_BYTES_LIMIT
當(dāng)broadcast join的數(shù)據(jù)量超過該閾值，則使用partition join
默認32G，是一個query option
對于有些集群，如果網(wǎng)絡(luò)是它的瓶頸，我們就可以考慮將這個值調(diào)小一點，限制廣播的數(shù)據(jù)量，對于我們的網(wǎng)絡(luò)負載是有一定幫助的。
② DISABLE_HDFS_NUM_ROWS_ESTIMATE
默認為false，表示當(dāng)hdfs表的統(tǒng)計信息缺失的時候，進行行數(shù)的預(yù)估
這就意味著這個表如果沒有統(tǒng)計信息的話，Impala會對它進行一個預(yù)估，不會像以前一樣非常暴力的采取一個默認的值。目前這個參數(shù)只對HDFS表有效，對于Kudu表是無效的。
5. 資源隊列配置模塊
在Impala2.12的時候，采用policy文件進行配置，需要修改每臺機器上的配置文件，新增或者修改一個隊列的話，就需要在所有的機器上把配置文件進行修改（可以只配置coordinator），這樣比較麻煩，不利于運維。基于這個情況，我們就開發(fā)了資源對列配置模塊。主要有以下這些功能：
頁面上新增/更新/刪除隊列
展示當(dāng)前的各個隊列信息
如圖所示：
1. 基于k8s和集群分組的動態(tài)伸縮
支持impala on k8s，實現(xiàn)Impala集群的快速部署和運維
基于集群分組，實現(xiàn)不同的業(yè)務(wù)隔離
支持集群節(jié)點的快速擴容/縮容
2. 基于Alluxio的HDFS文件緩存和優(yōu)化
實現(xiàn)Impala查詢緩存在Alluxio中的表
在Alluxio中實現(xiàn)分區(qū)級別的表緩存
支持query option，決定SQL查詢是否走Alluxio
實現(xiàn)小文件合并、文件格式轉(zhuǎn)換
3. 預(yù)計算與SQL路由
根據(jù)用戶配置或者SQL分析，通過物化視圖進行預(yù)計算
根據(jù)配置決定預(yù)計算的結(jié)果是否放在Alluxio中
用戶查詢路由，優(yōu)選選擇預(yù)計算的中間表
根據(jù)規(guī)則對SQL進行重寫，優(yōu)化SQL執(zhí)行
基于以上三個模塊，就有了如上圖這樣一個整體的架構(gòu)，這就是目前Impala在網(wǎng)易的一個整體的未來規(guī)劃。