基於Elasticsearch構建千億流量日誌搜索平台實戰

作者簡介：

王拓，七牛雲大數據高級工程師，碩士畢業於中國科學技術大學，曾就職於 PPTV 的廣告團隊，主要從事視頻廣告系統、ad exchange 交易平台、dsp 系統的設計和開發工作。於 2016 年加入七牛雲，主要負責日誌搜索雲服務 logdb 的架構和開發工作。目前 logdb 承載公司每天近千億、近百 TB 數據的日誌增量。

今天的分享會從五個方向展開：一，背景；二，系統的設計目標；三，我們在做這個系統過程中遇到的挑戰；四，如何應對；五，簡單的總結。

介紹這個系統之前，先介紹一下我們整個七牛雲 Pandora 大數據平台產品的形態圖。

這個產品形態圖最左邊會有數據源，用戶可以把數據導到 Pandora 大數據平台上，通過計算任務的加工，數據可以到一個消息隊列。這時候用戶可以選擇把數據導到日誌檢索，如果用戶有日誌檢索需求；

另外一條線可以把這個數據再導到新的計算任務，再到消息隊列裡面，這時候他可以選擇將數據導到時序資料庫，如果數據帶有時間戳的特徵。通過時序資料庫可以對數據進行一些報警或者監控分析；

另外也可以把數據導到七牛雲的對象存儲中，有什麼好處呢？對象存儲其實是七牛的存儲服務，這裡的定位和 hdfs 類似 ，用戶可以通過 Xspark 對已經存放在對象存儲里的數據進行多維數據分析。

如圖 5 所示是整個 Pandora 的系統架構圖。最左邊是接入層，我們可以通過 SDK 等方式把數據接入到 Pipeline 管道上，通過 transform 進行計算，這些 transform 就是剛剛對應的計算任務。然後可以通過 Export 導到指定的產品中。右邊可以通過 Xspark 等對它進行分析。

今天我們要討論的 Topic 是位於整個產品中的子產品，在 LogDB 的環節上。LogDB 是一個搜索雲服務。我們對這個產品的定位是這樣的，希望這個產品能夠基於 Pandora 的數據做一些分析服務，希望對普通用戶來講可以在 5—10 分鐘內可以完成整個數據的接入，同時我們希望每天可以搞定 MB 到 100TB 的日誌增量。當然對於用戶來說，最重要的是雲服務要做到 0 運維、0 開發、低成本。

我們對系統最初的設計目標如下：

1. 搞定公有雲的海量用戶，要支持各種規格；

2. 支持單個用戶日誌規模從 MB —100TB 每天；

3. 查詢秒級響應；

4. 系統必須要有可靠性，主要表現在兩方面，一方面不能丟數據，另外一方面系統可用性必須在 99.9% 以上；

5. 要擁抱開源，可以適配 Kibana、grafana 等。我們希望以前使用 ELK 的用戶可以用非常低的成本遷移到我們的上面，以前使用 ELK 的範式切過來，對你來說沒有任何變化。

有了設計目標之後，我們馬上要做出一個多租戶模型。

我們一開始就認為單個 ES 集群搞不定海量 repo ，因為大家都知道 ES 在 Master 被選舉之前是一個 P2P 的系統，但是當 Master 被選取後，它的管理本質上是 Master 和 slave 的模式。所以隨著集群規模的增大，以及 repo 數量的增多，整個集群管理負擔越來越大，對於 Master 要求會越來越高，最終會導致集群掛掉。

因為我們團隊都是比較有經驗的 ES 工程師，所以我們一開始就提出產品必須是多集群的海量 Cluster 的模型。如果 Cluster 出現資源不夠的情況，我們從運維的角度加一些新的 node 即可。另外如果某一個 Cluster 裡面 repo 達到了一定數量，超過了 ES 本身系統資源的瓶頸，我們就會選擇增加一個新的 Cluster 來搞定這個事情。

當我們敲定了多租戶模型之後很快迭代出了第一版系統，第一版系統架構圖如圖 7 所示，最上面一層會有 Portal、logical、SDK 等等。右邊是 export，我們的數據從 export 過來，所以要跟 export 對接。

在 API Server 這層我們做了一些服務的解藕，首先是一個 search 集群，一個索引集群，同時其他類型業務會被集中放在一個集群。我們的數據存在 Mongo 裡面，下面是整個 ES 集群的集合，最左邊需要有一個管理的服務，對 ES 集群進行維護，比如對 ES 進行優化等。

我們第一版系統上線之後遇到什麼問題？

首先，因為打點的 QPS 非常高，導致我們的 Mongo 成為系統里最先達到瓶頸的一環，它的查詢壓力非常大。那麼，怎麼解決這個問題？其實，我們需要一套 Cache 系統，這樣就可以把大量請求攔在 Mongo 之前來解決問題。不過還好，因為我們已經有了一套緩存系統，所以不需要花太長時間來解決這個問題。

七牛的緩存系統是一個二級緩存系統，第一級指的是 Local Cache，第二級是指 Memory Cache。舉個例子，當一個請求發出來，如果 Local Cache 沒有命中，它會去 Memory Cache 找，如果 Memory Cache 也沒有找到，這時候它會通過 qconf Master 到 Mongo 里找，如果 Mongo 找到之後，會把這個值回寫到 Memory Cache 和 Local Cache ，那麼第二次請求時直接在 Local Cache 會被攔掉。如果這個請求很不幸命中了另外一個API Server，實際上 Local Cache 是沒有的，但是 Memory Cache 也是有的，所以直接去 Memory Cashe 就可以拿到值了，然後再回寫 Local Cache。

通過計算，我們發現，基本上 99% 的請求都會被整個 Cache 系統攔住，而這個系統里基本上有 80% 的請求會被 Local Cache 攔住，也就是說整個 Memory Cache 和 Mongo 的壓力都是非常小的。所以我們馬上迭代出了新一版的系統，即我們加入了一個 Qconf Cluster。

這時候我們遇到的新問題是什麼？第一個問題是 LAG，經常有用戶反映這個系統查不到最近 10 分鐘或者 1 小時的數據。那麼，LAG 是怎麼產生的？我們認為整個寫點代碼是有問題的，寫點的整個效率有問題。所以，我們要對寫點做一些優化，在開始優化前，我們首先要了解 Benchmark，對整個效率有一個正確的認識。首先，我們要正確知道 ES 真正的吞吐量在哪裡，因為我們整體資源是夠的。比如，我們有十台機器，但是為什麼我們搞不定這些量，對用戶產生 LAG 了呢？所以我們先對 ES 集群做一個 Benchmark。

我們拿到一個具體的值，ES 集群做一個 Benchmark 要怎麼做？官方說，首先你一個並發控制 batchSize，從 0 加到 1 萬，當你的 batchSize 增加沒有任何收益時，固定這個值，再去增加它的並發數。當並發數增加也沒有收益的時候，固定的這兩個值就代表了 ES 最佳的效率。

那也就是說，其實 ES 打點的吞吐量其實取決於客戶端，對於這種推送系統其實大部分都是這樣。即我們對客戶端程序要求比較高，當然我們的數據如果是直接從內存裡面來的，沒有問題，我們可以一直維持這個並發數。但事實上我們的數據時 export 打給我們的，它發的請求可能是零散的，可能一個 batch 有可能一個只有一條數據或者一萬條數據，這種零散的數據打到 ES 之後，整個吞吐量會下滑。

所以，我們其實在 ES 之前需要做一個數據傳輸系統，這套系統要搞定 export 打給我們數據零散的點，我們要把它包裝好，以 ES 最佳的效率打給 ES，同時這套系統應該是多租戶的，我們內部稱這個系統叫 Producer。那麼，這個系統應該怎麼做？首先，我們對數據傳輸做一個簡單的思考，很多人經常會說，你做一個下游系統，對我的服務應該是穩定的，不管我以什麼樣的姿勢跟你交互。或者說傳輸的下游消費速度也是穩定的，很多時候我們也會認為，這個鏈路整體傳輸速度其實取決於上游和下游的影響。但事實上並非如此，對於上游或者下游的速度來講，吞吐量核心取決於並發數和並發大小，而維持最大吞吐量的核心是要維持並發數和並發大小。另外，要維持整體吞吐量要考慮三方面因素：拉取效率、鏈路效率和吞吐效率。

舉個例子，我們要搞定從 Kafka 到 ES 每秒 10K 的流量。我們知道 Kafka 效率很高，可能三個 Patation 可以搞定這個事情。那麼我們需要的並發是 3。它的整個batchSize 是 10K，這對 Kafka 是最友好的。但是對於 ES 來說，你要搞定每秒 10K 的流量，其實他的姿勢應該是這樣的，可能你的並發數是5，你的 batchSize 是 20K。

對於一個正常的系統，我們的 batchSize 該如何調？肯定是以最慢的為準，調到 20×5 的並發數，但事實上即便這樣也解決不了。因為鏈路本身也有效率的損耗，比如你數據是 JSON 格式，它首先要做 unMarshal，這個時間其實是 CPU 的開銷。所以真正要搞定這個問題，你可能需要大概 20K×8 的姿勢才能搞定這個問題。那麼，如何解決這個問題？我們認為我們需要引進一個東西來對上游下游做解藕，我們稱之為 Memory queue。

所以，我們首先需要一個隊列，而且是內存隊列，因為我們知道數據從上游拉下來之後進入內存隊列，它的效率是最高的。進入內存隊列之後，我們需要一個 Source 和 sink，因為我們要動態調整上游的 Source 數量和它的 batchSize 、下游的 sink 數量和它的 batchSize。與此同時，我們要做一個事務，為什麼呢？用 ES 的人都知道，當我們通過bulk介面打點的時候，一個請求裡面有 2w 個點，ES 其實是非常不友好的，他會告訴你這 2 萬個點裡有一半失敗一半成功。其實這對客戶端的負擔很重。所以我們希望這套系統帶事務，這樣對於客戶端來說，你打給我的點可以保證要麼都成功要麼都失敗。

我們基本模型做好之後，馬上迭代出了最基本的代碼。因為七牛是 Go 的深度用戶，所以我們很多系統都是用 Go 做的。在這個系統里，我們用 Go 的 channel 來搞定 memory queue。圖 9 是 channel 的一個生產者模型，首先你要事務池裡面拿到一個事務並開啟，同時要把數據塞給這個事務，然後把事務提交，把數據塞給 memory query，當然可能會失敗，要做回滾，回滾的時候是把數據從事務裡面刪掉，同時告訴用戶這次整體是失敗的。

圖 10 是 channel 的另外一端，是消費者的一個模型，首先你拿到事務，把數據從 channel 塞到事務里，然後嘗試去下游打點，如果成功了你可以提交事務，如果失敗了回滾事務。回滾是指把數據從事務回滾到 channal。

有了這麼一個基本模型之後，我們馬上要做的事情就是要把這個模型付諸於實踐，我們要先做一個單機版的 Producer。首先定義一個新的概念 task，task 由 source、sink 和 channel 組成。同時我們要構造一個新的概念 agent，agent 是對 task 進行管理。因為我們上游的 export 系統是通過 Http 協議來向我們發送數據的，所以我們需要一個 rest 的 source。有了這個基本系統後，我們還需要一個 checkpoint sink 來解決丟點的問題。因為我們的數據是在 memory queue 裡面，可能會有丟點的情況。

有了單機版的 Producer 後，馬上我們要解決一個新的問題，解決多租戶的 Producer，因為是我們公有雲的廠商。那麼如何解決呢？

圖 12 是多租戶的 Producer 模型。首先我們要對 task 進行基本描述，我們要描述它channel、batchSize 的大小，還有它的並發度。描述好 task 後，我們需要描述第二個東西就是 Producer agent metrics，它代表 cpu 資源、磁碟資源、網路資源，我們需要 Producer 對自己的所在機器的狀態能夠進行實時描述。有了這兩個基本點之後，我們馬上要做分布式系統，做分布式系統，我們首先想到的思路是通過 ZooKeeper 來解決，但是我們覺得用 ZooKeeper 來解決我們問題有點太重了，我們的服務是可以接受最終一致性的。所以我們覺得可以換一種新的思路去解決。

最終的思路如下：通過版本戳 + PULL 的模型搞定這個數據一致性問題。即通過定時到上游拉數據，由版本戳來判定這個數據是不是最新的，通過這種方式，整個系統的管理最終都會是一致的。

那麼這個架構就很明顯了，我需要一個 Master 對整個集群進行協調，同時我需要一個Producer agent，它表達的是資源單位。Producer agent 會定期把自己的狀態上報給 Master，Master 知道現在整個系統有多少個 task，它需要對 task 以及整體資源池作編排，我們叫它 rebalance。這樣每個 agent 就知道我要搞定多少 task，每個 task 狀態如何。

因為假如我們的 agent 非常多，對 Master 的壓力則非常大，所以用到了上文提到的 Qconf 集群來搞定 Cache 的問題。當然，我們相信任何自動化都是不靠譜的，總有 case 可能跑在已有的規則以外。所以這套系統剛開始設計的時候就增加了一個 admin 後台，我們可以對它進行人工干預。有了這套系統後，基本能保證我以最大的效率從 export 把數據導入到 ES。即我們通過這套系統搞定了 LAG。

緊接著，我們馬上會遇到一些新的挑戰，這個挑戰是什麼？就是大量的查詢超時。我們系統上線之後經常有用戶抱怨，譬如有用戶發現用在系統搜索十次，其中有八次都在超時，根本不可用。

怎樣解決這個問題？我們對線上數據做了一些採樣分析，然後發現以下規律：

• 用戶的 Query 總是多樣的，有可能一個 Query 掃的數據量只有一百萬，也可能是一個億、一百億、一千億。

• 搜索的過程其實有點像 mapper-reduce 的過程，比如 1 個數據有 1 個 shard，它搞定了 1 個 G 的量，在搜索的過程中如果是 10 毫秒沒問題。當數據增加的時候，可能需要 10 個搞定 10 個 G 的量，這時候每個的時間可能都是 10 毫秒，但是整體響應時間其實大於 10 毫秒，因為這個時候做 reduce 的節點（es 叫 coordinating node）要 merge 的數據量更大。也就是隨著 shard 的增減，它的搜索體驗會越來越差，即當客戶嘗試搜索 1 千億條數據，它可能落在 100 個 shard 上，那麼它的搜索響應時間是不可估算的。

聽起來這個問題很難去解，我們覺得問題是很複雜的，但是我們可以抓住其中一些主要的變數，那這是什麼？我們首先還是要做 Benchmark，這樣才能正確認識到瓶頸到底在哪裡。

譬如我們測了一個 shard ，他搞定數據，實際的 QPS 是多少。這時候 QPS 是很重要的關鍵點，我們可以控制 QPS。比如每個用戶的 Query 是很複雜的，我可以控制 QPS 對限制某個用戶對資源的掠奪，這樣可以保證整個 shard 的 QPS 不會出現特別高的情況，是在可控範圍之內。另外要對日誌的 Query 進行優化，針對特定的 Query 調優會比對 es 參數的調優收到比顯著的效果。

所以我們的思路是：首先通過 QPS 控制整體的量，然後再做 Query 的優化，把一些大的 Query 拆成小的 Query。

那麼如何優化？首先看一下日誌 Query 的特點，在我們看來日誌 Query 天生帶有時間的 Tag，另外它的排序是時間排序，不像一個廣告系統的排序可能是相關度。有了這兩點之後我們就可以展開了。

舉個例子，一個用戶 Query 來了之後，它是帶時間的，我們知道這個時間所需要的 Shard 大概在哪一天，這樣我們可以避免無用的資源消耗，相當於是把大的 Query 拆成小的 Query，因為之前我們的 search是要掃所有的 shard。假如我確定它在第一天，這時我到對應的 shard 上搜即可。還有就是滿足條件預判呢？比如，它對這次搜索的結果希望是反饋 200 條數據，那因為你是時間排序，很有可能你去搜索的時候第一個Shard 就已經有 200 條數據了，這時候我們可以提前返回。

第二個思路是，我們要做一個執行計劃的可控。那麼，怎麼去控制一個執行計劃？我們剛才有講到，搜索體驗隨著 Shard 的增多而增多。同時搜索 100 個 Shard ，它的體驗非常差，所以我們要控制。經過測試發現 5 個 Shard 的搜索體驗最好。

我們的執行計劃是這麼做的，比如你要搜索最近一年的數據，那怎樣去做呢？如果按照之前的姿勢，你會把整個系統資源佔掉，把系統拖垮，還拿不到數據。而現在我們在後台把這個做成串列的，先給你搜索 5 個 Shard，5 個 Shard 一直疊加，而用戶看到的是一個進度條，通過這種方式，用戶更容易接受，不會感到茫然。因為我們之前的搜索時間經常到 60 秒或者 60 秒開外，這時候用戶會感到茫然，說我等了這麼長時間你給我看一個 503。

我們其實關於 Query Excecutor 這塊的優化點其實蠻多，我挑了關鍵的三個點講一下。搞定了查詢的超時問題似乎大部分問題搞定了，但其實也不是這樣，我們的挑戰永遠都存在。我們經常會在晚上 24 點被用戶叫來，說你們的集群掛掉了，我們經常凌晨 2、3 點一起討論到底什麼原因導致的。

關於 24 點掛掉的原因，用 ES 的人可能都清楚，ES 的自帶的 TTL 機制其實很笨，效率非常差。那我們要避免要使用這種姿勢，按天建立索引，所以不可避免在 24 點會有一些問題，比如到 24 點都要對它進行創建索引，而且這個創建索引非常扎堆，扎堆是指比如現在有 1000 用戶，可能需要 2000 個 shard 搞定這個事情，也就是說在 24 點這一刻馬上創建 2000 個 Shard，這個對於整個集群的壓力是非常大的，因為整個任務會持續堵塞，會造成整個集群內存爆炸，最終導致 Master 掛掉，集群解散。

這個問題解法其實非常簡單，我們可以提前一天平滑創建。24 點的問題核心是扎堆創建，所以我們要平滑，我們可以提前一天的 3 點一個一個去創建索引，保證整個集群的壓力是可控的。

這個事情聽下來還是有點玄乎，因為感覺都不太好控制，所以我們還是根據之前的解決方案，抓大頭，只要搞定寫點和查詢即可。之前已經強調過寫點的 batchSize ，查詢的 QPS 我們都是有的。那我們可以考慮對 shard 進行切分，70% 的資源用來搞定寫點，30% 的資源用來搞定查詢。這時候整個 shard 都是一個可量化的指標，譬如，這個 Shard 可以搞定多少寫點，這個 Shard 可以搞定多少查詢。有了 shard 這個標準化概念後，我們要馬上要搞定它。

但搞定它之前，應該先看看現在面臨的問題是什麼樣？我們的系統是基於 ES 構建的，但是它 rebalance 演算法是有局限性的。用過 ES 的人都知道，它 rebalance 的考慮因子主要有以下幾個方面：

第一，對磁碟的使用率，第二，認為每一個 Shard 都是一樣的，他不知道隨著時間的推移每一個 Shard 能搞定的量是多少。按天的方式，第一天我們有 10 個節點 10 個Shard，每個節點搞定一個 Shard。但是經過一天的發展，每個節點上的 Shard 數據量量都不一樣。這時候到了第二天，某幾個節點的磁碟使用率可能非常低，這 10 個 Shard 有可能出現互相的堆積，集中在其中 3—4 台節點上。也就是說我們不能依靠 ES 搞定這個事情，我們需要自己動手。

再看看另外一邊，我們能夠正確描述一個 Shard，因為 Shard 編排是雙向的，首先你要表達清楚需要多少資源，然後才能根據我的現有資源進行安排。但是在另外一邊，用戶打點的量和他查詢的 K8S ，其實都是彈性的，因為你不知道公有雲用戶什麼時候來我們平台，你根本不知道他的流量需求、查詢需求是多少。如果他是來查問題的。比如查一下日誌，當然這種 QPS 非常低，大概每天幾百個。他有可能用你的服務構建一個系統，這種系統寫點可能比較低，查詢比較高。所以現在首先 ES 不能幫我們搞定這個事情，我們需要自己動手搞。其次，用戶流量不可預期。

那麼，怎麼解這個問題？我們認為需要從四個方面考慮：

首先是冷啟動。對於一個新用戶來到我們平台之後，我們對他其實是沒有任何認知的，這時候我們需要給他一個冷啟動的標配。比如，他來之後給他一個默認的標配，給他分一定量資源。

另外要做流量預估，對於一個已經在我們平台上跑了很多天的用戶，他的流量是可預期的。我們可以根據他歷史上多少天對流量進行預估，這跟廣告比較像，廣告裡面有一個庫存的概念，你要提前預估庫存才能進行售賣，同理，要預估流量來採購資源，保證系統平滑運行。這套系統現在是在跑在 XSpark 上的。

另外需要一個實時擴容縮容的東西。什麼意思呢？就是當某用戶是在我們平台上跑了大概三個月，表現非常好，流量很穩定。但如果是一個電商用戶，比如他 618 大促，這種我們演算法搞不定，它跑在我們的演算法之外。所以我們需要進行動態的擴容縮容，它不依賴於任何演算法去做預測，只是根據現有的情況，實時增加容量或者減少容量，它是基於我們之前的 Pandora workflow 做的，我們也不需要做基建，只需要搞定這個系統就可以了。

搞定了流量預估、動態擴縮容，我們應該不會存在熱點了。這裡面還有一個問題需要注意，就是穩定性的事情。就是我們的編排演算法一定要穩定，這個穩定是什麼意思？

之前我們在介紹的 Producer 的時候大家也注意到，Producer 是一個輕狀態的server，它的遷移成本非常低，而對於 ES Shard ，遷移成本非常高，比如這個 Shard 本來已經搞定了 100G 的量，這時候屬於演算法不穩定，每天演算法跑完之後編排結果出來之後，需要對 Shard 進行大量挪動遷移，其實對整個集群壓力非常大。因為你想 1 個 T 的 Shard 如果從這個磁碟拷到另外一個磁碟，大致也是需要好幾個小時的。

所以這個演算法必須是穩定，必須能夠以最小的 Shard 編排結果搞定熱點問題。搞定了 Shard 編排系統，又搞定了 Producer 系統，還搞定了搜索的查詢執行計劃系統，看似我們的系統應該是完美了。

因為我們團隊 ES 工程師有經驗的比較多，所以我們覺得可以用 ES 搞定這個問題。但是在項目選型之初，我們覺得不希望系統被 ES 綁定，所以我們的系統其實是跟ES 解耦的，包括 API 的一些介面定義都是跟 ES 無關的，也就是我們希望我們的系統後面能掛到更多的引擎，事實上我們現在在搞定之前一些比較頭疼的問題之後，我們現在也在嘗試一種新的可能，嘗試讓我們的系統可以掛在更多的引擎解決這個問題。

我說完那麼多，那麼，我們到底搞定了多少事情？

第一，支撐海量用戶；第二，支撐每天100T+、2000億+ 的流量；第三，沒有 LAG；第四，查詢可以保證秒級返回；第五，接近 0 運維，因為即便我們系統現在做的特別好，但是總有一些不在我們的現在系統能夠做的範圍之內，這個指的是編排等分配問題，所以有時候需要人工去干預一下；第六，現在系統能夠做到可用性在 99.9%。

最後再給大家做個預告，七牛雲 Pandora 大數據平台內測邀請中，點擊閱讀原文即可參與。

喜歡這篇文章嗎？立刻分享出去讓更多人知道吧！