高小青：Impala在神策數(shù)據(jù)實(shí)時(shí)分析引擎中的落地與優(yōu)化｜DataFunTalk

導(dǎo)讀：今天分享的主題是“Impala落地與優(yōu)化：神策數(shù)據(jù)多維分析平臺(tái)構(gòu)建實(shí)戰(zhàn)”，分為五個(gè)部分：

• 神策產(chǎn)品技術(shù)架構(gòu)
• 基于Impala的實(shí)時(shí)分析引擎
• 查詢性能優(yōu)化
• 查詢資源預(yù)估
• 未來(lái)計(jì)劃

01

神策產(chǎn)品技術(shù)架構(gòu)

1. 神策數(shù)據(jù)產(chǎn)品架構(gòu)

這是整個(gè)的產(chǎn)品架構(gòu)圖，分為三個(gè)部分，第一部分是數(shù)據(jù)根基，其次是營(yíng)銷(xiāo)云和分析云。數(shù)據(jù)根基部分分為采集、傳輸、治理、存儲(chǔ)、查詢，以及數(shù)據(jù)智能幾個(gè)部分，底層有我們的私有云平臺(tái)。在分析云部分，除了已經(jīng)升級(jí)了的用戶行為分析、指標(biāo)預(yù)警、用戶畫(huà)像等，還有最新研發(fā)的廣告投放分析以及經(jīng)營(yíng)數(shù)據(jù)分析，我們希望能給用戶提供一個(gè)完整的分析體驗(yàn)。其次就是營(yíng)銷(xiāo)云，現(xiàn)在已經(jīng)提供了完整的運(yùn)營(yíng)活動(dòng)、微信生態(tài)運(yùn)營(yíng)，以及流程畫(huà)布，我們希望能夠打造一個(gè)用戶體驗(yàn)的產(chǎn)品的數(shù)據(jù)閉環(huán)。我們?cè)谏蠈右矔?huì)提供一個(gè)數(shù)據(jù)驅(qū)動(dòng)的咨詢服務(wù)，來(lái)幫助用戶更好的完善自己的數(shù)據(jù)分析體驗(yàn)。

2. 神策數(shù)據(jù)技術(shù)架構(gòu)

接下來(lái)介紹一下神策數(shù)據(jù)技術(shù)架構(gòu)。

首先最左側(cè)黃色的部分是各個(gè)導(dǎo)入的SDK，包括服務(wù)端SDK、客戶端SDK，以及導(dǎo)入工具比如LogAgent、Batch Importer等。

通過(guò)Nginx進(jìn)入到日志文件接收系統(tǒng)，Extractor會(huì)對(duì)文件進(jìn)行解析，包括一些校驗(yàn)、處理、清洗等等，最后生成復(fù)合神策協(xié)議的規(guī)范的文件，進(jìn)入到Kafka數(shù)據(jù)系統(tǒng)。

Data Loader是神策數(shù)據(jù)自研的一個(gè)數(shù)據(jù)接收系統(tǒng)，它會(huì)實(shí)時(shí)訂閱Kafka中的數(shù)據(jù)，將這些數(shù)據(jù)實(shí)時(shí)寫(xiě)入到Kudu中，并且在一段時(shí)間內(nèi)有定時(shí)子任務(wù)將Kudu中數(shù)據(jù)轉(zhuǎn)存到Parquet文件格式中，保證了實(shí)時(shí)寫(xiě)入并且利用了列存的優(yōu)勢(shì)能夠快速的讀取。

上層還有Yarn的任務(wù)調(diào)度系統(tǒng)，它會(huì)定時(shí)的調(diào)度一些Kafka的消費(fèi)任務(wù)，以及預(yù)處理的一些任務(wù)。

接下來(lái)就是基于Impala構(gòu)建的實(shí)時(shí)查詢引擎，在上層客戶會(huì)傳給Query Engine一個(gè)可以解析的Request，查詢引擎Query Engine會(huì)將它翻譯成業(yè)務(wù)可理解的SQL，查詢引擎會(huì)給它返回結(jié)果，同時(shí)在前端展示子系統(tǒng)中也有自己的緩存，保證能夠降低Impala的一些壓力。另外整個(gè)系統(tǒng)底層還有Monitor可以進(jìn)行監(jiān)控。

02

基于Impala的實(shí)時(shí)分析引擎

1. 用戶行為需求

隨著時(shí)間的增大，維度越來(lái)越多，并且維度的取值也非常分散，但是我們又希望能夠滿足客戶各種維度下鉆的需求，可以看出我們的查詢模式是非常多樣化的。其次就是我們要能實(shí)時(shí)響應(yīng)客戶的需求，并且它的查詢頻率是較低的。所以用戶分析構(gòu)建是將靈活性排在第一位，其次是及時(shí)性，最后是時(shí)效性。

2. Impala架構(gòu)特點(diǎn)

首先Impala是基于MPP查詢引擎的，它的計(jì)算和存儲(chǔ)是在同一臺(tái)節(jié)點(diǎn)上，并且共享自己的內(nèi)存、磁盤(pán)、CPU等，這樣每個(gè)節(jié)點(diǎn)可以方便的進(jìn)行并行計(jì)算。

Impala包含3個(gè)進(jìn)程：StateStore進(jìn)程、CatalogD進(jìn)程、ImpalaD進(jìn)程。StateStore進(jìn)程主要負(fù)責(zé)監(jiān)控集群里各個(gè)節(jié)點(diǎn)的ImpalaD的健康狀態(tài)、實(shí)時(shí)接收ImpalaD進(jìn)程的注冊(cè)訂閱消息，CatalogD進(jìn)程通過(guò)Hive Metastore去緩存元數(shù)據(jù)信息，如查詢需要用到的表信息、各種屬性字段信息，以及每個(gè)查詢需要用到的Parquet文件地址，將這些數(shù)據(jù)緩存起來(lái)，并且可以通過(guò)Statestore將這些緩存的元數(shù)據(jù)分發(fā)給各個(gè)Impalad節(jié)點(diǎn)，每個(gè)Impalad維護(hù)一份自己的緩存數(shù)據(jù)。Impalad又分為兩種角色，一種是Coordinator角色負(fù)責(zé)接收請(qǐng)求以及匯總查詢結(jié)果反饋給前端，一種是Executor角色負(fù)責(zé)執(zhí)行計(jì)劃?？梢钥闯鲭m然Impala有較高的內(nèi)存需求，但是它的查詢效率是非常高的，這也是我們選擇它的重要原因。針對(duì)它較低的容錯(cuò)性和較高的內(nèi)存需求問(wèn)題，我們會(huì)在后續(xù)迭代中進(jìn)一步完善。

3. 基于Impala的系統(tǒng)架構(gòu)

這是基于Impala的系統(tǒng)架構(gòu)。各種SDK以及導(dǎo)入工具，首先將數(shù)據(jù)寫(xiě)入Kudu，最終轉(zhuǎn)換成Parquet文件格式中，底層支持用戶行為表、用戶表，同時(shí)也支持客戶自己導(dǎo)入的如維度表，以及系統(tǒng)也支持一些Iterm表給用戶自己定義。針對(duì)上層Impala來(lái)說(shuō)，它的用戶行為表是將Kudu和HDFS上的數(shù)據(jù)Union起來(lái)的，所以它不需要再去拷貝數(shù)據(jù)，它看到的是一張視圖。在上層，我們也做了一些查詢緩存，來(lái)防止Impala有過(guò)大的壓力。

03

查詢性能優(yōu)化

接下來(lái)介紹一下我們的查詢優(yōu)化，包括五個(gè)部分，分別是舊存儲(chǔ)模式、新存儲(chǔ)優(yōu)化、基于用戶行為序列的查詢優(yōu)化、外連接消除優(yōu)化和預(yù)處理表達(dá)式優(yōu)化。其中部分功能已經(jīng)提交到社區(qū)。

1. 舊存儲(chǔ)模式

說(shuō)到查詢優(yōu)化，不得不先說(shuō)一下存儲(chǔ)優(yōu)化，首先介紹下舊的存儲(chǔ)模式。

舊的存儲(chǔ)模式是數(shù)據(jù)按天、按Event進(jìn)行分區(qū)，同時(shí)每個(gè)區(qū)的數(shù)據(jù)文件大小都有一定的規(guī)則，保證最優(yōu)的掃描效率；其次它的數(shù)據(jù)是部分有序的。但是現(xiàn)在這個(gè)存儲(chǔ)也面臨著一些問(wèn)題，針對(duì)這里提到的三類(lèi)場(chǎng)景的優(yōu)化，沒(méi)有達(dá)到非常好的效果。

• 首先是在復(fù)雜分析的場(chǎng)景中，數(shù)據(jù)是需要按用戶以及查詢的事件時(shí)間是完全有序的，但是因?yàn)榇鎯?chǔ)不是全局有序的，后面需要對(duì)上億甚至幾十億條數(shù)據(jù)做全排序。內(nèi)存不夠的情況下，會(huì)落到磁盤(pán)上，這樣查詢就會(huì)非常慢。
• 第二點(diǎn)就是有一些事件是高頻查詢，會(huì)經(jīng)常用到，但是有一些事件可能是客戶導(dǎo)入的，并不需要實(shí)時(shí)查，如兩年以前的數(shù)據(jù)，如果放在HDFS上并不是有非常好的效果。
• 第三點(diǎn)是有一些事件是需要頻繁更新的，比如說(shuō)最近兩天的訂單狀態(tài)（已購(gòu)買(mǎi)、已配送等）是會(huì)實(shí)時(shí)更新的，用HDFS的Parquet格式存，沒(méi)有辦法做到這一點(diǎn)。

2. 新存儲(chǔ)優(yōu)化

針對(duì)上面三類(lèi)場(chǎng)景，我們進(jìn)行了一些優(yōu)化。

在新的存儲(chǔ)模式下，

• 首先，它是按照天、用戶ID、時(shí)間進(jìn)行預(yù)排序，保證底層存儲(chǔ)格式的更進(jìn)一步的有序性。
• 第二點(diǎn)就是針對(duì)一些不需要實(shí)時(shí)查詢的數(shù)據(jù)，如兩年前的數(shù)據(jù)，我們支持客戶自定義的歸檔策略，把它放到AWS S3這種性價(jià)比較高的存儲(chǔ)上，幫助客戶節(jié)約成本。
• 第三點(diǎn)是針對(duì)一些需要實(shí)時(shí)更新的事件，我們可以單獨(dú)把它存儲(chǔ)到Kudu中，可以用戶自定義什么時(shí)候把它放到HDFS上，更加的有靈活性，支持各種各樣的場(chǎng)景。

3. 基于用戶行為序列的查詢優(yōu)化

在剛才存儲(chǔ)格式進(jìn)行優(yōu)化后，我們接下來(lái)看如何針對(duì)復(fù)雜查詢也就是基于用戶行為序列的查詢，怎么進(jìn)行優(yōu)化。

首先看左邊的查詢計(jì)劃。首先底層間進(jìn)行掃描，每個(gè)節(jié)點(diǎn)掃描出一部分?jǐn)?shù)據(jù)，接下來(lái)在這個(gè)節(jié)點(diǎn)上做Union，比如將HDFS或Kudu上的數(shù)據(jù)進(jìn)行匯總。匯總后，各個(gè)節(jié)點(diǎn)會(huì)進(jìn)行1個(gè)Exchange，exchange后會(huì)將相同用戶ID分發(fā)到相同的Impalad節(jié)點(diǎn)，然后會(huì)進(jìn)行一個(gè)全排序，最終再到ETL后算出一個(gè)結(jié)果?？梢钥吹?，這可能會(huì)排序幾億甚至幾十億的數(shù)據(jù)量，這是非常消耗內(nèi)存資源的，也是非常慢的，是很大的瓶頸。因此，可以利用底層掃描數(shù)據(jù)的有序性，在上層Exchange里做一個(gè)Shuffle exchange，保證給到上層節(jié)點(diǎn)的數(shù)據(jù)都是按用戶、按時(shí)間排好序的，這樣我們就可以干掉sort節(jié)點(diǎn)，直接進(jìn)入U(xiǎn)DTF的算子中。可以看出，我們直接對(duì)這個(gè)查詢計(jì)劃進(jìn)行了一個(gè)優(yōu)化。

最后我們?cè)谶@樣一個(gè)測(cè)試集群（規(guī)模是10個(gè)節(jié)點(diǎn)，每個(gè)節(jié)點(diǎn)32g內(nèi)存4核CPU，非SSD的磁盤(pán)，大概模擬了30天的真實(shí)數(shù)據(jù)，每天約5億條數(shù)據(jù)）上對(duì)計(jì)劃進(jìn)行了分析。可以看出，我們對(duì)不同的分析模型不同的天進(jìn)行了測(cè)試，基本上有6倍到40倍的提升，內(nèi)存會(huì)降為之前的1/5。比如7天的3步漏斗，之前大概需要30秒左右，現(xiàn)在基本上能把它變到10秒以內(nèi)，這個(gè)效果還是非常明顯的，并且已經(jīng)推給了客戶升級(jí)。

4. 外連接消除優(yōu)化

接下來(lái)要講的是一個(gè)已經(jīng)提交給社區(qū)的外連接消除優(yōu)化。

比如針對(duì)上圖中的SQL，將Event和Profile表連接起來(lái)，加了左表和右表的屬性過(guò)濾，并且屬性是非null的。本來(lái)是使用Full Join的，但是在這種場(chǎng)景下，可以轉(zhuǎn)換成Inner Join。轉(zhuǎn)成Inner Join后，可以利用它自帶的將B表的ID構(gòu)建1個(gè)哈希傳遞給左表，左表構(gòu)建一個(gè)類(lèi)似于Runtime Filter的數(shù)據(jù)結(jié)構(gòu)，這樣在Join之后，分發(fā)給上層的算子的數(shù)據(jù)量會(huì)大幅度減少。

5. 預(yù)處理表達(dá)式優(yōu)化

這個(gè)SQL也是從漏斗的SQL中截取的一部分。比如在內(nèi)層會(huì)對(duì)一些事件還有屬性做過(guò)濾得到一個(gè)Funnel_step_ID，在上層的聚合中會(huì)用到這個(gè)ID。通過(guò)預(yù)處理表達(dá)式，復(fù)雜的表達(dá)式case when是在底層的Scan層去做的，Scan層是多線程的。如果沒(méi)有做優(yōu)化，則在上層的union層做表達(dá)式計(jì)算，效率非常低。

下圖展示的是我們的優(yōu)化：

首先，針對(duì)漏斗的例子，本來(lái)是在Scan層，Scan層上再Union層計(jì)算復(fù)雜表達(dá)式，并且把所有數(shù)據(jù)全部傳遞給上層，但是在優(yōu)化之后，進(jìn)行下推，把case when或者正則匹配的過(guò)濾在Scan層做，這樣Union的時(shí)候就已經(jīng)不需要傳遞多余的屬性，直接把最終列的屬性傳遞給上層，這樣就可以減少數(shù)據(jù)發(fā)送量。其次，可以利用Scan的多線程的操作模式，大大提升效率。

在上面講的測(cè)試集群上進(jìn)行測(cè)試，針對(duì)漏斗使用的列數(shù)不同，它需要發(fā)送的數(shù)據(jù)量會(huì)大幅減少，基本上能減少80%以上。比如漏斗里，需要城市（如北京）、訂單金額（如大于200元）、訂單狀態(tài)（如成功）這樣3列過(guò)濾，本來(lái)需要把3列的數(shù)據(jù)都Scan出來(lái)發(fā)送到上層，但是優(yōu)化之后，不需要發(fā)送3列，直接算出最終的那一列，數(shù)據(jù)量減少到3.4G，效果是非常明顯的。

04

查詢資源預(yù)估

1. 現(xiàn)狀與解決對(duì)策

首先我們可以分析一下，平時(shí)我們?cè)谶\(yùn)維過(guò)程中遇到的查詢問(wèn)題。一類(lèi)主要是查詢資源不足，還有一類(lèi)是查詢慢。查詢慢主要通過(guò)上面提到的性能優(yōu)化來(lái)解決，針對(duì)查詢資源不足這種錯(cuò)誤率較高的情況，原因主要是兩類(lèi)。

一類(lèi)是資源預(yù)估的內(nèi)存不準(zhǔn)。比如預(yù)估的內(nèi)存太小了，而集群上沒(méi)有那么多資源了，因此查詢時(shí)會(huì)報(bào)內(nèi)存不足；而如果預(yù)估的資源太大，并發(fā)度又不夠。所以如何提高資源內(nèi)存預(yù)估的準(zhǔn)確性，是我們減少錯(cuò)誤率的重中之重。針對(duì)內(nèi)存預(yù)估不準(zhǔn)，我們給出了三種解決方案。第一種是基于歷史查詢資源的預(yù)估，首先會(huì)將歷史的資源根據(jù)各個(gè)算子生成簽名，存到k-v存儲(chǔ)系統(tǒng)中，后續(xù)有類(lèi)似查詢過(guò)來(lái)時(shí)，根據(jù)各個(gè)算子、時(shí)間條件、過(guò)濾條件等，進(jìn)行一個(gè)等比例的放大縮小。另外一種是歷史上沒(méi)有類(lèi)似的查詢，就使用公式資源預(yù)估的方式。Impala自己帶一個(gè)公式預(yù)估的方式，但是不是非常準(zhǔn)確，偏差比較大。我們主要針對(duì)常用的Agg、Join、Sort這三個(gè)算子內(nèi)存進(jìn)行公式預(yù)估的優(yōu)化，效果還是比較明顯。第三種是Impala是沒(méi)有容錯(cuò)機(jī)制的，如果這次查詢失敗，不會(huì)再次進(jìn)行查詢，我們這里是給它一次機(jī)會(huì)，讓它查詢失敗后重新預(yù)估內(nèi)存，再去試一次。

另外一類(lèi)是大查詢會(huì)阻塞小查詢，這里主要是分為大小查詢隊(duì)列，保證小查詢?cè)谝粋€(gè)隊(duì)列，大查詢?cè)诹硗庖粋€(gè)隊(duì)列。其次我們有改進(jìn)后的基于時(shí)間的調(diào)度算法，Impala自帶的是先入先出的算法，能夠按序去調(diào)度查詢，我們會(huì)保障后繼小查詢會(huì)根據(jù)算法能夠及時(shí)的進(jìn)行查詢。

2. 查詢資源預(yù)估流程

首先是查詢過(guò)來(lái)后，根據(jù)查詢計(jì)劃生成簽名。比如針對(duì)漏斗有Scan、Union、Exchange、再到上層的UDTF，我們會(huì)結(jié)合這些算子，以及每個(gè)算子中的信息，比如天數(shù)、過(guò)濾條件、上層的Join條件等等信息，生成簽名，然后判斷歷史否有這樣的簽名，如果有則根據(jù)歷史進(jìn)行等比例的放大縮小去計(jì)算內(nèi)存，如果沒(méi)有則根據(jù)Impala自帶的公式預(yù)估，進(jìn)入到查詢調(diào)度里面。然后用自己實(shí)現(xiàn)的調(diào)度算法，讓它執(zhí)行，執(zhí)行成功就會(huì)返回結(jié)果，并且更新歷史上存到k-v庫(kù)中的簽名以及對(duì)應(yīng)的耗時(shí)、內(nèi)存。如果執(zhí)行失敗，會(huì)進(jìn)行重試，再次進(jìn)入查詢調(diào)度器讓它執(zhí)行，如果還是失敗，則最終失敗。這是整個(gè)的預(yù)估流程。

3. 查詢資源預(yù)估效果

因?yàn)轭A(yù)估的準(zhǔn)確性，可以很大程度上決定了錯(cuò)誤率降低的效果。這也是我們?cè)趧偛诺臏y(cè)試集群上，分別對(duì)10種不同分析模型、查詢時(shí)長(zhǎng)算出的資源預(yù)估的效果。

其中藍(lán)色表示每個(gè)查詢實(shí)際需要的內(nèi)存、綠色是基于歷史相似查詢預(yù)估的內(nèi)存、灰色是我們根據(jù)Impala自帶的公式預(yù)估進(jìn)行優(yōu)化后算出的內(nèi)存。可以看出，除了極個(gè)別灰色差別比較大主要是帶Join情況下優(yōu)化有一定誤差，其它情況下我們的歷史預(yù)估、基于公式預(yù)估與實(shí)際內(nèi)存相比較都是非常接近的。這個(gè)我們給四五十家客戶上線，可以將錯(cuò)誤率降低80%以上，效果是非常明顯的。

05

未來(lái)計(jì)劃

最后，介紹下我們的未來(lái)計(jì)劃。

首先，上面的一些功能，比如內(nèi)存優(yōu)化，以及資源預(yù)估，有一部分已經(jīng)推回社區(qū)了，但是還有很大一部分沒(méi)有推回社區(qū)。后面會(huì)將這些功能進(jìn)行拆分推回社區(qū)，來(lái)提升Impala性能。

另一點(diǎn)，是我們最近也在做的彈性計(jì)算和查詢可觀測(cè)性。其中彈性計(jì)算是我們集群的動(dòng)態(tài)的縮容擴(kuò)容，節(jié)約客戶成本，提升查詢體驗(yàn)。查詢可觀測(cè)性是我們最近在做的客戶管理自己查詢資源的一個(gè)系統(tǒng)，希望能夠做到客戶自己在發(fā)現(xiàn)查詢慢的時(shí)候，可以根據(jù)查詢可觀測(cè)性系統(tǒng)，去動(dòng)態(tài)發(fā)現(xiàn)一些大查詢是否有用動(dòng)態(tài)管理自己的查詢資源，以及管理自己的導(dǎo)入資源。

最后，我們會(huì)持續(xù)做性能優(yōu)化，保證業(yè)界先進(jìn)水平。

06

精彩問(wèn)答

Q: 有序漏斗分析如何做？

A: 如果問(wèn)漏斗怎么做，可是用UDTF做，主要是一些實(shí)現(xiàn)邏輯。如果問(wèn)優(yōu)化怎么做，主要還是要對(duì)底層的數(shù)據(jù)進(jìn)行優(yōu)化，保證scan出來(lái)的數(shù)據(jù)已經(jīng)是按用戶、按時(shí)間有序的了，這樣各個(gè)節(jié)點(diǎn)在上層再去做歸并，保證一個(gè)用戶在同1個(gè)節(jié)點(diǎn)上是完全有序的了，省去一個(gè)Sort算子，達(dá)到一個(gè)優(yōu)化效果。應(yīng)該是在各個(gè)系統(tǒng)里都比較相似。

Q: /*materialize_expr*/是你們自研的嗎？

A: 對(duì)，是我們自研的一個(gè)hint，后續(xù)會(huì)把它做成自動(dòng)識(shí)別，而不需要業(yè)務(wù)端加hint，會(huì)把它推回社區(qū)。

Q: 具體推回社區(qū)的特性有哪些呢？

A: 第一個(gè)推回社區(qū)的特性是外鏈接消除，已經(jīng)分批次推回社區(qū)了。第二部分是復(fù)雜表達(dá)式下推，這個(gè)我們也會(huì)盡快推回社區(qū)。因?yàn)樯鐓^(qū)也會(huì)有一些任務(wù)，我們也會(huì)跟社區(qū)合作去做。其次就是有一些優(yōu)化的，也會(huì)拆分出好多點(diǎn)。比如公式預(yù)估的準(zhǔn)確性優(yōu)化，已經(jīng)優(yōu)化自帶的FIFO的調(diào)度算法優(yōu)化，都可以拆出很多點(diǎn)推回社區(qū)。

Q: Kudu到HDFS的數(shù)據(jù)排序，是在Impala里實(shí)現(xiàn)的嗎？

A: 在導(dǎo)入邏輯里實(shí)現(xiàn)的，Impala主要是用這個(gè)數(shù)據(jù)。導(dǎo)入的邏輯是我們自研的Dataloader系統(tǒng)實(shí)現(xiàn)的，主要是先把數(shù)據(jù)寫(xiě)入到Kudu中，其次會(huì)定期對(duì)Kudu數(shù)據(jù)進(jìn)行轉(zhuǎn)Parquet的實(shí)現(xiàn)。

今天的分享就到這里，謝謝大家。