Impala 在 Hulu 中的優化和改進

摘要： 點選hadoop123關注我喲 知名的大資料中臺技術分享基地，涉及 大資料架構(hadoop/spark/flink等) ， 資料平臺(資料交換、資料服務、資料治理等) 和 資料產品(BI、AB測試平臺) 等，也會分享 最新技術進展，大資料相關職位和求職資訊，大資料技術交流聚...

點選hadoop123 關注我喲

知名的大資料中臺技術分享基地，涉及 大資料架構 (hadoop/spark/flink等) ， 資料平臺(資料交換、資料服務、資料治理等) 和 資料產品(BI、AB測試平臺) 等，也會分享 最新技術進展，大資料相關職位和求職資訊，大資料技術交流聚會、講座以及會議 等。

背景

Impala 是一個 SQL on Hadoop 的 MPP 查詢引擎，由 Cloudera 主導開發並捐獻給 Apache 軟體基金會，在 2017 年底正式孵化成為 Apache 頂級專案。

如圖，大資料領域裡的 OLAP 系統種類繁多，它們各自有自己的特長和侷限性，需要根據實際業務場景合理進行選擇。儲存方式往往決定了系統的能力和相容性，跟據儲存方式我們可以把它們分為三大類：

1. 不儲存原始資料，只保留聚合結果，如 Druid 、 Kylin 等

2. 儲存計算一體，自己實現儲存層，如 Greenplum 、 Clickhouse 、 Doris （原 Baidu Palo ）等

3. 儲存計算分離，依託 HDFS 、 S3 等實現儲存層，這類系統又可以細分為兩類：

1. 使用自定義的檔案格式，如 HAWQ 、 VectorH 等

2. 使用開原始檔格式，如 Hive 、 Impala 、 Presto 、 SparkSQL 、 Drill 等

第 1 類系統不保留原始資料，可以把效能做得很高，但由於聚合方式需要事先定義，比較適合報表類業務等查詢模式相對固定的場景。當需要明細查詢或互動式的靈活查詢（ Adhoc 查詢）時，仍需要另兩類系統的加入才能支援。

第 2 類系統自己管理資料，可以做很多定製的優化，從而也能達到較高的效能。但由於資料沒法與其它系統共享，往往需要將已有資料重新匯入一次，在引入時需要考慮遷移成本以及多存一份資料的代價，因此一般在新建數倉中被考慮使用。第 3.a 類系統雖然依託外部儲存，但因為採用封閉的檔案格式，仍需要多存一份資料，其地位其實與第 2 類相同，常常在新建數倉時才被考慮。

使用最廣的還是屬於 3.b 類的系統，主要原因是基於 Hadoop 的數倉使用廣泛，此類系統不需要重新生成一份資料，能與已有架構充分相容。在此類系統中， Impala 、 Presto 、 Drill 屬於 MPP （ Massively Parallel Processing ）系統，各節點流式地在記憶體中完成計算，中間資料幾乎不落盤，相比 Hive 、 SparkSQL 等批處理系統能達到極高的效能。

在 Hadoop 生態圈的 MPP 系統中， Impala 具有優異的效能，它的優勢緣於：

• C++ 實現，相比 Presto 、 Drill 等 Java 實現要更高效，也省去了 Java GC 的開銷

• 基於 LLVM 的 Code Generation ，能根據實際資料型別生成高效的執行程式碼

• CBO (Cost Based Optimizer) ，基於各表的統計資訊得出代價最低的執行計劃

• 快取元資料，生成執行計劃時不再需要和 Hive 、 HDFS 進行互動

• RuntimeFilter ，在執行時基於已讀的小表資料裁剪大表需要掃描的資料量

• Predicate/Aggregation Pushdown ，雖然 MPP 系統都會有下推優化，但支援的程度各有不同，也依賴於底層檔案格式的支援。 Impala 在 Parquet 儲存格式上做了很多 native 的下推優化，如 IMPALA-3654 、 IMPALA-4624 、 IMPALA-4985 、 IMPALA-6113 等。

Impala在Hulu的數倉中 有很多應用 ，我們對 Impala 做了一些核心級別的開發和優化，希望能與大家共同探討。

Impala內部原理

Impala 叢集由一臺 Catalog Server ( 簡稱 catalogd) ，一臺 Statestore Server ( 簡稱 statestore) 和若干 Impala Daemon ( 簡稱 impalad) 組成。自 Impala 2.10 之後， Impalad 又可分為 Coordinator 或 Executor 兩種角色。各服務的功能如下：

• Statestore 負責管理叢集心跳和廣播元資料更新。

• Catalogd 負責從 Hive 和 HDFS 拉取元資料並快取下來，同時將元資料更新發給 statestore 進行廣播。另外 catalogd 還負責執行建表、新建 partition 等 DDL 語句。

• Impalad 中的 Coordinator 也會快取元資料，負責接收 SQL 查詢請求並生成執行計劃，並將執行計劃的分片（ PlanFragment ）排程到各 Executor 去執行，最終彙總結果返回給客戶端。

• Impalad 中的 Executor 則只負責 PlanFragment 的執行。

下圖展示了一個 SQL 查詢的執行過程：

(1) 只有 Coordinator 角色的 Impalad 才會接收客戶端請求。 Coordinator 對查詢進行語法分析、語義分析。

(2) 語義分析中需要查詢各表的元資料（元資料的具體內容見後文），如果在該 Coordinator 的元資料快取中缺失，則會向 catalogd 請求載入。 catalogd 會向 Hive Metastore 和 HDFS NameNode 查詢所要的元資料，並將元資料的更新發送給 statestore 進行廣播，從而所有 Coordinator 都會得到更新。

(3) Coordinator 通過語義分析生成執行計劃，並根據資料的本地性（ locality ）將 Plan Fragment 排程到各 Executor 去執行

(4) Executor 從 HDFS 讀取資料，並將 PlanFragment 例項的執行結果返回給上層結點，最終彙總到 Coordinator 得到最終結果。

Impala 的執行計劃遵從 Volcano 的 Iterator 模型，是由若干 PlanNode 組成的執行計劃樹，葉子節點讀取外部資料並傳輸給上層節點做下一步處理，最終在根節點彙總。

Iterator 模型使得整個查詢可以最大限度地流式進行，從而降低了查詢的延遲。 Volcano 模型的另一大貢獻是引入了 Exchange 節點，使得執行計劃可以劃分為不同的分片，各自採用合適的並行度去執行。

如圖是 TPC-H Benchmark 中 Query3 的執行計劃，根據是否需要 broadcast 、 shuffle 等被切分成幾個 PlanFragment 。每個 Executor 執行的就是 Plan Fragment 的例項。

Hulu對Impala的改進

1. 增加對ORC檔案格式的支援

ORC 是一種列式儲存的檔案格式，由 Hortonworks 主導開發，而 Cloudera 主推的是 Parquet 。因此在 Cloudera 和 Hortonworks 宣佈合併之前， Impala 並沒有支援 ORC 的計劃。由於歷史原因， Hulu 的 Hive 數倉中大量使用了 ORC 儲存格式，為了引入 Impala ，我們決定對它進行核心級的修改，讓其支援 ORC 儲存格式。我們分兩步走，第一步先實現基本型別（ primitive types ）的支援（ IMPALA-5717 ），第二步再增加了巢狀型別（ struct 、 array 、 map ）的支援（ IMPALA-6503 ）。這兩部分工作均已貢獻給社群， impala 在 2.12 及 3.1 版本開始支援讀取 ORC 檔案中基本型別的列，在 3.2 版本支援讀取 ORC 檔案中巢狀型別的列。

這部分工作的核心是實現一個 HdfsOrcScanner ，因為 Query 執行的大部分邏輯如語法分析、語義分析、排程等基本可以複用已有的實現，唯獨最終解析 ORC 檔案這塊需要專門的實現。前面我們介紹過了 Impala 的執行計劃樹，樹的葉子節點都是 ScanNode 。每個 ScanNode 的例項負責讀取若干個資料分片（ split ），每個 split 由一個 Scanner 執行緒去處理。如下圖所示，編號 02 的 HdfsScanNode 有 14 個例項，分別執行在 14 個 Executor 上。每個例項會啟動若干個 Scanner 執行緒來讀取 split 。

Impala 支援的所有 HDFS 檔案格式（如 Parquet 、 Avro 、 SequenceFile 、 RCFile 、 Text 等）都有一個對應的 scanner 實現，為了支援 ORC ，我們同樣要實現一個 HdfsOrcScanner 。

上圖是 HdfsOrcScanner 的內部結構，主要可分為以下幾方面：

(1) Impala 如何管理記憶體 ： Impalad 會追蹤每個查詢佔用了自己多少記憶體，超過閾值的查詢會被 kill 掉。 HdfsOrcScanner 的記憶體管理要遵從既有的流程，從而讓 impalad 能正確統計記憶體佔用量（通過 impala::MemTracker ）。

(2) Impala 如何讀取資料 ： Scanner 並不需要真正讀取 HDFS 上的資料， Impala 把 IO 讀取封裝成了 DiskIoMgr 。 ORC 檔案的讀取並不是從頭讀到尾，而是先解析檔案尾得到元資訊，然後跳到每個 Stripe （行組）中讀取所需的列。每個 Stripe 的讀取又要先解析 Stripe 尾部的元資訊。這些都要求 Scanner 正確地與 DiskIoMgr 進行互動，避免無用的 IO 。

(3) Impala 如何表示資料 ：不管底層檔案是列存還是行存， Scanner 都會將其物化（ materialization ）成為記憶體中按行存放的 Tuple ，若干個 Tuple 組成 RowBatch 返回給 ScanNode 。每個 Tuple 包含了一行中被選擇的各列資料，具體的樣子由 TupleDescriptor 進行描述。 Scanner 需要理解 TupleDescriptor ，並將 ORC 資料物化成所需的 Tuple 。這塊的工作比較細，比如需要考慮 Tuple 所引用的記憶體空間的生命週期管理、 TupleDescriptor 所要的列在 ORC 檔案中是否存在及是否相容、遇到正常中斷（如被 cancel 或達到 limit ）或解析異常時的處理等。

(4) 如何解析 ORC 格式的檔案 ： ORC Reader 已經有 C++ 版的官方開源實現，我們直接將其封裝在 HdfsOrcScanner 裡即可，主要的工作是把前 3 個層面封裝成 ORC Reader 的引數或輸入，並解析 ORC Reader 的輸出。在整合 ORC Reader （屬於 ORC library ）的過程中，我們還發現並修復了一些 bug ，詳見 ORC-311 、 ORC-312 、 ORC-313 、 ORC-314 、 ORC-317 、 ORC-403 。

上圖對比了 Impala on Parquet 、 Impala on ORC 、 Presto on ORC在TPC-H基準測試中的20個查詢的耗時（單位：秒）。可以看到Impala on ORC的效能雖然比不上Impala on Parquet，但相比Presto on ORC還是有很大的優勢。 Impala的ORC scanner還有很多優化可做，比如支援Aggregation Pushdown、結合ORC檔案的統計資訊來減少無用IO、使用DiskIoMgr的非同步IO介面等，理論上應該能達到與Parquet scanner相近的效能。關於後續的工作，歡迎關注IMPALA-6505、IMPALA-6636、IMPALA-8046等相關JIRA。

2. 自動重新整理元資料

Impala 快取了 Hive 中各表的元資料，包括列的定義、 partition 的位置和許可權、 HDFS 檔案的資訊（大小、許可權、複本位置等）。這些資訊從 Hive Metastore (HMS)  和 HDFS NameNode (NN) 得來，當查詢再次訪問相同的表時， Impala 可以利用快取的元資料直接生成執行計劃並開始執行，省去了對 HMS 和 NN  的多次互動。

這是 Impala 元資料層的設計初衷，確實加速了查詢效能，也降低了對 HMS 和 NN 的訪問壓力，但因此引入了兩個非常不友好的語句:  "INVALIDATE METADATA"  和  "REFRESH" 。

當 Hive 中的表有更新時（如新增 partition 或重新覆蓋了原表資料）， Impala 並不能自動感知，需要使用者手動執行 " REFRESH tableName"  語句來重新整理元資料快取。如果在 Hive 中建了一個新表，還需要在 Impala 中執行 " INVALIDATE METADATA tableName"  來通知 Impala 這個新表的存在。如果沒有及時操作，對應表上的查詢基本都會掛掉。

為了將 Impala 無縫引入我們已有的 Hadoop 數倉，我們需要將元資料重新整理自動化。如圖所示，我們搭建了一個 pipeline ，當 Hive 中的表有更新時， Hive MetaStore 會記下一條 audit 日誌。 Audit collector 將其傳送到 Kafka ，然後被一個 Flink job 消費，觸發 Impala 重新整理快取。

這條 pipeline 可以達到秒極延遲，但維護起來還是有點麻煩。幸運的是， Impala 在 3.2 版本引入了自動重新整理元資料的功能（ IMPALA-7970 ），將來也會 merge 到 2.x 的版本中去。自動重新整理元資料的功能還有許多細化的工作，具體見 IMPALA-7954 。這部分工作是 Cloudera 和 Hortonworks 合併之後才開始的，還處於起步階段，大家可以關注一下。

3. Built-in的get_json_object函式  

get_json_object 是 Hive 中一個處理 JSON 字串的函式，用於抽取 JSON 中的指定內容。 Impala 中並沒有該函式的 native 實現，我們需要將 Hive 中實現該函式的 Java 類定義成 Impala 的 UDF 才能使用。在低版本的 Hive （ apache 版本小於 2.3 或 cdh 版本小於 5.12.0 ）中這個函式有記憶體洩漏的 bug （ HIVE-16196 ），而且 Impala 目前還沒法追蹤 各查詢在JVM 裡 所佔用的記憶體，我們的 Impala 叢集曾因此遭遇了 OOM 。為此我們實現了 native 的 get_json_object 函式，並貢獻給了社群 (IMPALA-376) 。

總結

Impala 是 SQL-on-Hadoop 中一個高效能的 MPP 查詢引擎。本文簡要介紹了 Impala 的內部原理，以及 Hulu 在實際應用中對其做的一些優化和改進，包括增加對 ORC 檔案格式的支援、外圍的自動重新整理元資料框架、支援 native 的 JSON 處理函式等。

大家在使用 Impala 中遇到的任何問題，歡迎加入Impala技術交流群 與我們探討！同時也歡迎加入 Impala 社群的 SlackChannel（文末有連結） ！

如果二維碼過期，請在微信公眾號中回覆”impala“獲取最新二維碼

參考文獻與連結

• Cloudera Document, "Impala RuntimeFilter"”, www.cloudera.com/documentation/enterprise/latest/topics/impala_runtime_filtering.htm

• GoetzGraefe, " Encapsulation of Parallelism in the Volcano Query ProcessingSystem", http://daslab.seas.harvard.edu/reading-group/papers/encapsulation-volcano.pdf

• Impala Slack Channel, https://join.slack.com/t/apache-impala/shared_invite/enQtNTgzMzAyNzIyNTk0LTQwMzJjMDI0YzEwOWRmZDk2MzNlZTk5OWZkNTI4M2Y5MmU1MjQ1ZWIzYzQxMWQyMjUzNjNjNWU0NDQ1MTMyNWM

• "ORC Specification v1", https://orc.apache.org/specification/ORCv1/

• Mostafa Mokhtar, "Performance Optimizations inApache Impala", https://www.slideshare.net/cloudera/performance-of-apache-impala

作者簡介：黃權隆，中國第一個Apache Impala PMC成員，畢業於北大計算機系網路所資料庫實驗室，目前就職於Hulu大資料基礎架構團隊，主要專注於OLAP引擎相關技術。