Impala源碼之資源管理與資源隔離
本文由 網易雲 發布。
前言
Impala是一個MPP架構的查詢系統，為了做到平臺化服務，首先需要考慮就是如何做到資源隔離，多個產品之間盡可能小的甚至毫無影響。對於這種需求，最好的隔離方案無疑是物理機器上的隔離，A產品使用這幾臺機器，B產品使用那幾臺機器，然後前端根據產品路由到不同集群，這樣可以做到理想中的資源隔離，但是這樣極大的增加了部署、運維等難度，而且無法實現資源的共享，即使A產品沒有任務在跑，B產品也不能使用A產品的資源，這無疑是一種浪費。毛主席教導我們浪費是可恥的，所以我們要想辦法在充分利用資源的情況下實現產品之間的資源隔離，這其實是一個非常有難度的工作。
YARN
在大數據生態圈，談到資源管理（Resource Management）和隔離（Resource Isolation），第一反應想到的肯定是YARN，它是自Hadoop2.0開始並且一直使用的一種資源管理系統， YARN主要通過集中式的資源管理服務Resource Manager管理系統中全部的資源（主要是CPU和內存），然後為每一個產品或者業務定義一個隊列，該隊列中定義了提交到該隊列下的任務最大申請的資源總量；當一個任務提交到Resource Manager之後，它會啟動一個ApplicationMaster 來負責該任務的資源申請和調度，然後根據任務需要的資源需求，向Resource Manager申請資源，Resource Manager 根據當前隊列中資源剩余情況判斷是否授予資源，如果當前隊列資源已經被用盡則該任務需要等待直到有資源釋放，等到ApplicationMaster申請到資源之後則請求NodeManager啟動包含一定資源的Container，Container利用cgroups輕量級的隔離方案實現，為了根據不同的使用場景YARN也集成了不同的分配和調度策略，典型的有Capacity Scheduler和Fair Scheduler。
上圖展示了客戶端提交任務到YARN的流程，平時在提交MR、spark任務時也是通過這種方式，但是對於MPP架構的系統，它的查詢響應時間由最慢的節點運行時間決定，而為了提升查詢性能，又需要盡可能多的節點參與計算，而YARN上的任務每次都是啟動一個新的進程，啟動進程的時間對於批處理任務是可以接受的，畢竟這種任務運行時間比較久，而對於追求低延遲的Ad-hoc查詢而言代價有點大了，很可能出現進程啟動+初始化時間大於真正運行的時間。
除了使用原生的yarn調度，impala也嘗試過使用一個稱之為Llama（Long-Lived Application Master）的服務實現資源的管理和調度，它其實是YARN上的一個ApplicationMaster，實現impala和yarn之間的協調，當一個impala接收到查詢之後，impala根據預估的資源需求向Llama請求資源，後者向YARN的Resource Manager服務申請可用的資源。但是前面提到了，impala為了保證查詢速度需要所有的資源同時獲得，這樣才能推進下一步任務的執行，實際上，Llama實現了這樣的批量申請的功能，所以一個查詢的進行需要等到一批資源同時到達的時候才能夠進行下去，除此之外，Llama還會緩存申請到的資源。但是Llama畢竟還是需要從YARN申請資源並且啟動進程，還是會存在延遲比較大的問題，因此，Impala也在2.3版本之後不再支持Llama了。
Impala資源隔離
目前Impala的部署方式仍然是啟動一個長時間運行的進程，對於每一個查詢分配資源，而在新版本（2.6.0以後），加入了一個稱為Admission Control的功能，該功能可以實現一定意義上的資源隔離，下面我們就深入了解一下這個機制，看一下它是如何對於資源進行隔離和控制的。
首先，如果根據impala的架構，所有的SQL查詢，從解析、執行計劃生成到執行都是在impalad節點上執行的，為了實現Admission Control，需要在impalad配置如下了兩個參數：
1. --fair_scheduler_allocation_path 該參數是用來指定fair-scheduler.xml配置文件路徑，該文件類似於YARN的fair- scheduler.xml配置，具體配置內容下面再詳細講述；
2. --llama_site_path 該參數用來指定Llama的配置文件llama-site.xml，上面不是說到新版本不用Llama了嗎？為什麽還要配置它呢，其實這裏面的配置項都是一些歷史遺留項了吧。
接下來就詳細介紹一下如何配置這兩個文件，對於第一個文件fair-scheduler.xml，熟悉YARN的都知道該文件實現公平調度器的配置，YARN中的公平調度是如何實現的我不懂，但是文件中基本上需要配置一下每一個隊列的資源分配情況，下面是一個配置實例：
<queue name="sample_queue">
<minResources>10000 mb,0vcores</minResources>
<maxResources>90000 mb,0vcores</maxResources>
<maxRunningApps>50</maxRunningApps>
<weight>2.0</weight>
<schedulingPolicy>fair</schedulingPolicy>
<aclSubmitApps>charlie</aclSubmitApps>
</queue>
但是通過impala源碼發現impala中用到的每一個隊列的配置只有aclSubmitApps和maxResources，前者用於確定該隊列可以由哪些用戶提交任務，如果用戶沒有該隊列的提交權限（隊列中沒設置），或者用戶沒指定隊列則提交到default隊
列，如果default隊列不存在或者用戶沒有提交到的default隊列的權限，則拒絕該請求；後者是用於確定該隊列在整個集群中使用的最大資源數，目前impala關註的資源只有memory。在上例中sample_queue隊列在整個集群中能夠使用的內存大小是90GB，只有charlie用戶能夠提交到該隊列。
既然只用到這兩個配置，為什麽impala不單獨搞一個配置格式呢，而選擇直接用fair-schedular.xml呢？我想一方面是為了省去自己寫解析類了，直接使用yarn的接口就可以了，另外為以後更加完善做準備。下面再看一下Llama配置中用到了什麽配置，配置實例如下：
<property>
<name>llama.am.throttling.maximum.placed.reservations.root.default</name>
<value>10</value>
</property>
<property>
<name>llama.am.throttling.maximum.queued.reservations.root.default</name>
<value>50</value>
</property>
<property>
<name>impala.admission-control.pool-default-query-options.root.default</name>
<value>mem_limit=128m,query_timeout_s=20,max_io_buffers=10</value>
</property>
<property>
<name>impala.admission-control.pool-queue-timeout-ms.root.default</name>
<value>30000</value>
</property>
這些配置的意義如下，具體的配置項則是如下key後面再加上隊列名：
//隊列中同時在跑的任務的最大個數，默認是不限制的llama.am.throttling.maximum.placed.reservations
//隊列中阻塞的任務的最大個數，默認值是200 llama.am.throttling.maximum.queued.reservations
//隊列中阻塞的任務在阻塞隊列中最大的等待時間，默認值是60s
Impala實現
好了 ， 分析完了Admission Control中使用的配置項，用戶可以在創建一個session之後通過set REQUEST_POOL=pool_name的方式設置改session的請求提交的隊列，當然如果該用戶沒有該隊列的提交權限，之後執行都會失敗。下面根據查詢的流程看一下impala如何利用這些參數完成資源隔離的 當impala接收到一個查詢請求之後，請求除了包含查詢SQL之外，還包括一批查詢參數，這裏我們關心的是該請求提交的隊列（REQUEST_POOL參數），它首先根據查詢執行的用戶和隊列參數獲得該查詢應該提交到的隊列，選取隊列的規則如下：
1. 如果服務端沒有配置fair-scheduler.xml和llama-site.xml，說明沒有啟動資源控制服務，則所有的請求都提交到一個名為default-pool的默認隊列中；
2. 如果該查詢沒有指定REQUEST_POOL，則將REQUEST_POOL設置為yarn默認隊列default。
判斷隊列名是否存在，然後再根據當前提交任務的用戶和隊列名判斷該用戶是否具有提交任務到隊列的權限。如果隊列名不存在或者該用戶無權限提交則查詢失敗。
查詢執行完初始化工作（選擇隊列只是其中的一部分工作）之後會調用FE的GetExecRequest接口進行執行計劃的生成， 生成執行計劃的流程大致分為三部：
1. 語法分析生成邏輯執行計劃並進行預處理；
2. 根據邏輯執行計劃生成單機執行計劃；
3. 將單機執行計劃轉換成物理自行計劃，後續再單獨介紹這部分。
接著impalad節點會根據執行計劃判斷該查詢是否可以繼續執行，只有出現如下幾種情況時，查詢需要排隊：
1. 當前隊列中已經有查詢在排隊，因為阻塞隊列是FIFO調度的，所以新來的查詢需要直接排隊；
2. 已經達到該隊列設置的並發查詢上線；
3. 當前查詢需要的內存不能夠得到滿足。
前兩種條件比較容易判斷，對於第三種情況，impala需要知道當前查詢需要的內存和當前隊列中剩余的內存情況進行判斷，這裏的內存使用分為兩個方面：集群中隊列剩余的總內存和單機剩余內存。首先判斷隊列中剩余內存和當前查詢需要在集群中使用的內存是否達到了隊列設置的內存上限；然後在判斷該查詢在每一個impalad節點上需要的內存和該節點剩余內存是否達到設置的內存上限（節點的內存上限是由參數mem_limit設置的）。那麽問題又來了，該查詢在整個集群需要多少內存，在每一個節點上需要多少內存是如何計算的呢？對於整個集群上需要內存其實就是每一個節點需要的內存乘以需要的節點數，那麽核心問題就是該查詢需要在每一個節點使用的內存大小。
可能大家和我一樣覺得，對於每一個查詢需要在每一個節點消耗的內存是根據查詢計劃預估出來的，但是這樣做是非常難的，那麽來看一下當前impala是如何做的，對於單節點內存的預估，按照如下的優先級計算查詢需要的單機內存：
1. 首先判斷查詢參數rm_initial_mem是否被設置（可以通過set rm_initial_mem =xxx設置），如果設置了則直接使用該值作為預估值；
2. 然後判斷impalad啟動的時候是否設置rm_always_use_defaults=true，如果設置了則使用rm_default_memory中配置的內存大小；
3. 接著再判斷該session是否設置了mem_limit（可以通過set mem_limit=xx設置，註意它和impalad啟動時的mem_limit配置的區別），如果設置則使用該值；
4. 最後根據判斷執行計劃中是否計算出了每一個節點需要分配的內存大小；
5. 如果以上都沒有命中則使用默認的rm_default_memory配置（impalad啟動時候的參數），該值默認值是“4GB”。
從上面的的判斷邏輯來看，impalad最後才會根據執行計劃中預估的值確定每一個節點分配的內存大小，畢竟只是根據統計信息預估出來的信息並不是準確的，對於一些復雜的查詢而言，可能是誤差非常大的。
好了，通過上面的分析，整個查詢審計流程梳理了一遍，但是如果當前資源不能夠滿足該查詢呢？此時就需要將該查詢放入隊列中，該查詢則會阻塞直到滿足如下兩種條件之一：
1. 該查詢所使用的隊列擁有了該查詢需要的足夠的資源；
2. 該查詢存放在隊列中並超時，超時時間是由隊列中的queue_timeout_ms參數設置，如果隊列沒設置則由impalad啟動時的queue_wait_timeout_ms參數決定，默認是60s。
再談Statestore
我們這裏就不深究執行計劃中的內存估計是如何計算的了，但是還有一個比較重要的問題：每一個impalad是獨立工作的，只有在需要分配任務的時候才會通知其余的impalad執行相應的operation，那麽impalad如何知道其他impalad節點的資源狀態的？包括每一個隊列已使用的內存大小，每一個節點已使用的內存大小等。這就靠我們上一篇文章中介紹的statestored了，在statestored中impalad啟動時會註冊一個impala-request-queue主題，每一個impalad都是該topic的發布者同時也是訂閱者，周期性的發布當前節點的內存使用情況，然後每一個impalad節點再根據topic中最新的信息更新整個集群的資源使用狀態。這種交互方式的確挺方便，但是可能存在一定的不確定性，例如某一次impalad與statestored 的網絡抖動都有可能導致無法獲取到最新的資源使用狀態。
硬性限制
impalad使用Admission Control實現了一定意義上的資源隔離，但是這畢竟是軟性的，而不是像YARN那種通過cgroup 啟動新進程來進行隔離，仍然有可能存在一個比較繁重的查詢將整個集群搞垮，對於這種情況作為查詢平臺我們需要做到即使返回錯誤也不能影響這個整個集群的服務，此時就需要祭出impala查詢的一個關鍵性參數：mem_limit，對於impalad中的每一個模塊，啟動時都可以設置mem_limit參數，該參數是每一個執行節點能夠分配的最大內存（tcmalloc 管理），而每一個查詢也可以設置mem_limit，它表示該查詢在每一個節點上最大分配的內存大小，再每一個impalad執行查詢（impala中稱之為fragment）的時候會分配一個大小為mem_limit的block pool，需要的內存從pool中分配並保存在內存中，如果分配的內存超出pool的大小，則選擇一定的block spill到外存中，這部分具體的執行流程也是非常復雜了，可以單獨再講，這部分block在需要的時候再從本地磁盤讀取到內存中，需要spill到外存的查詢無疑會拖慢查詢速度， 但是這很好的保存了整個系統的可用性和穩定性。
總結
在實際部署的時候，我們會根據每一個用戶的數據量大小、業務類型分配隊列，甚至相同的業務不同時間區間的查詢分配不同的隊列，並且詳細設置該隊列中的默認查詢參數，尤其是mem_limit參數和最大並發數，這樣可以較好的限制租戶之間的影響，為了避免惡意用於的使用，可以限制用戶自己設置MEM_LIMIT參數，盡可能得保證集群的穩定性。
網易有數：企業級大數據可視化分析平臺。面向業務人員的自助式敏捷分析平臺，采用PPT模式的報告制作，更加易學易用，具備強大的探索分析功能，真正幫助用戶洞察數據發現價值。可點擊這裏免費試用。
了解 網易雲 ：
網易雲官網：https://www.163yun.com/
新用戶大禮包：https://www.163yun.com/gift
網易雲社區：https://sq.163yun.com/
Impala源碼之資源管理與資源隔離