百億級實時查詢優化實戰，讓你的Elasticsearch飛起來！

最近的一個項目是風控過程數據實時統計分析和聚合的一個 OLAP 分析監控平臺，日流量峯值在 10 到 12 億上下，每一年數據約 4000 億條，佔用空間大概 200T。

面對這樣一個數據量級的需求，咱們的數據如何存儲和實現實時查詢將是一個嚴峻的挑戰。

通過對 Elasticsearch 多方調研和超過幾百億條數據的插入和聚合查詢的驗證以後，咱們總結出如下幾種可以有效提高性能和解決這一問題的方案：

1. 集羣規劃
2. 存儲策略
3. 索引拆分
4. 壓縮
5. 冷熱分區等

本文所使用的 Elasticsearch 版本爲 5.3.3。

什麼是時序索引？其主要特色體如今以下兩個方面：

**存，**以時間爲軸，數據只有增長，沒有變動，而且必須包含 timestamp（日期時間，名稱隨意）字段。

其做用和意義要大於數據的 id 字段，常見的數據好比咱們一般要記錄的操做日誌、用戶行爲日誌、或股市行情數據、服務器 CPU、內存、網絡的使用率等。

**取，**必定是以時間範圍爲第一過濾條件，而後是其餘查詢條件，好比近一天、一週、本月等等，而後在這個範圍內進行二次過濾。

好比性別或地域等，查詢結果中比較關注的是每條數據和 timestamp 字段具體發生的時間點，而非 id。

此類數據通常用於 OLAP、監控分析等場景。

1、集羣部署規劃

咱們都知道在 Elasticsearch（下稱 ES）集羣中有兩個主要角色：Master Node 和 Data Node，其餘如 Tribe Node 等節點可根據業務須要另行設立。

爲了讓集羣有更好的性能表現，咱們應該對這兩個角色有一個更好的規劃，在 Nodes 之間作讀取分離，保證集羣的穩定性和快速響應，在大規模的數據存儲和查詢的壓力之下可以坦然面對，各自愉快的協做。

一、Master Nodes

Master Node，整個集羣的管理者，負有對 index 的管理、shards 的分配，以及整個集羣拓撲信息的管理等功能。

衆所周知，Master Node 能夠經過 Data Node 兼任，可是，若是對羣集規模和穩定要求很高的話，就要職責分離，Master Node 推薦獨立，它的狀態關乎整個集羣的存活。

Master 的配置：

node.master: true

        node.data: false

        node.ingest: false
複製代碼

這樣 Master 不參與 I、O，從數據的搜索和索引操做中解脫出來，專門負責集羣的管理工做，所以 Master Node 的節點配置能夠相對低一些。

另外防止 ES 集羣 split brain（腦裂），合理配置 discovery.zen.minimum_master_nodes 參數，官方推薦 master-eligible nodes / 2 + 1 向下取整的個數。

這個參數決定選舉 Master 的 Node 個數，過小容易發生「腦裂」，可能會出現多個 Master，太大 Master 將沒法選舉。

二、Data Nodes

Data Node 是數據的承載者，對索引的數據存儲、查詢、聚合等操做提供支持。

這些操做嚴重消耗系統的 CPU、內存、IO 等資源，所以，應該把最好的資源分配給 Data Node，由於它們是真正幹累活的角色，一樣 Data Node 也不兼任 Master 的功能。

Data 的配置：

node.master: false

        node.data: true

        node.ingest: false
複製代碼

三、Coordinating Only Nodes

ES 自己是一個分佈式的計算集羣，每一個 Node 均可以響應用戶的請求，包括 Master Node、Data Node，它們都有完整的 Cluster State 信息。

正如咱們知道的同樣，在某個 Node 收到用戶請求的時候，會將請求轉發到集羣中全部索引相關的 Node 上，以後將每一個 Node 的計算結果合併後返回給請求方。

咱們暫且將這個 Node 稱爲查詢節點，整個過程跟分佈式數據庫原理相似。那問題來了，這個查詢節點若是在併發和數據量比較大的狀況下，因爲數據的聚合可能會讓內存和網絡出現瓶頸。

所以，在職責分離指導思想的前提下，這些操做咱們也應該從這些角色中剝離出來，官方稱它是 Coordinating Nodes，只負責路由用戶的請求，包括讀、寫等操做，對內存、網絡和 CPU 要求比較高。

本質上，Coordinating Only Nodes 能夠籠統的理解爲是一個負載均衡器，或者反向代理，只負責讀，自己不寫數據。

它的配置是：

        node.master: false

        node.data: false

        node.ingest: false

        search.remote.connect: false
複製代碼

增長 Coordinating Nodes 的數量能夠提升 API 請求響應的性能，咱們也能夠針對不一樣量級的 Index 分配獨立的 Coordinating Nodes 來知足請求性能。

那是否是越多越好呢？在必定範圍內是確定的，但凡事有個度，過了負做用就會突顯，太多的話會給集羣增長負擔。

在作 Master 選舉的時候會先確保全部 Node 的 Cluster State 是一致的，同步的時候會等待每一個 Node 的 Acknowledgement 確認，因此適量分配可讓集羣暢快的工做。

search.remote.connect 是禁用跨集羣查詢，防止在進行集羣之間查詢時發生二次路由：

2、Routing

相似於分佈式數據庫中的分片原則，將符合規則的數據存儲到同一分片。ES 經過哈希算法來決定數據存儲於哪一個 Shard：

shard_num = hash(_routing) % num_primary_shards
複製代碼

其中 hash(_routing) 得出一個數字，而後除以主 Shards 的數量獲得一個餘數，餘數的範圍是 0 到 number_of_primary_shards - 1，這個數字就是文檔所在的 Shard。

Routing 默認是 id 值，固然能夠自定義，合理指定 Routing 可以大幅提高查詢效率，Routing 支持 GET、Delete、Update、Post、Put 等操做。

如：

PUT my_index/my_type/1?routing=user1

{

"title": "This is a document"

}

GET my_index/my_type/1?routing=user1
複製代碼

不指定 Routing 的查詢過程：

簡單的來講，一個查詢請求過來之後會查詢每一個 Shard，而後作結果聚合，總的時間大概就是全部 Shard 查詢所消耗的時間之和。

指定 Routing 之後：

會根據 Routing 查詢特定的一個或多個 Shard，這樣就大大減小了查詢時間，提升了查詢效率。

固然，如何設置 Routing 是一個難點，須要一點技巧，要根據業務特色合理組合 Routing 的值，來劃分 Shard 的存儲，最終保持數據量相對均衡。

能夠組合幾個維度作爲 Routing ，有點相似於 Hbase Key，例如不一樣的業務線加不一樣的類別，不一樣的城市和不一樣的類型等等，如：

        _search?routing=beijing：按城市。

        _search?routing=beijing_user123：按城市和用戶。

        _search?routing=beijing_android，shanghai_android：按城市和手機類型等。
複製代碼

數據不均衡？假如你的業務在北京、上海的數據遠遠大於其餘二三線城市的數據。

再例如咱們的業務場景，A 業務線的數據量級遠遠大於 B 業務線，有時候很難經過 Routing 指定一個值保證數據在全部 Shards 上均勻分佈，會讓部分 Shard 變的愈來愈大，影響查詢性能，怎麼辦？

一種解決辦法是單獨爲這些數據量大的渠道建立獨立的 Index

這樣能夠根據須要在不一樣 Index 之間查詢，然而每一個 Index 中 Shards 的數據能夠作到相對均衡。

另外一種辦法是指定 Index 參數 index.routing_partition_size，來解決最終可能產生羣集不均衡的問題，指定這個參數後新的算法以下：

shard_num = (hash(_routing) + hash(_id) % routing_partition_size) % num_primary_shards
複製代碼

index.routing_partition_size 應具備大於 1 且小於 index.number_of_shards 的值。

最終數據會在 routing_partition_size 幾個 Shard 上均勻存儲，是哪一個 Shard 取決於 hash(_id) % routing_partition_size 的計算結果。

指定參數 index.routing_partition_size 後，索引中的 Mappings 必須指定 _routing 爲 "required": true，另外 Mappings 不支持 parent-child 父子關係。

不少狀況下，指定 Routing 後會大幅提高查詢性能，畢竟查詢的 Shard 只有那麼幾個，可是如何設置 Routing 是個難題，可根據業務特性巧妙組合。

3、索引拆分

Index 經過橫向擴展 Shards 實現分佈式存儲，這樣能夠解決 Index 大數據存儲的問題。

但在一個 Index 變的愈來愈大，單個 Shard 也愈來愈大，查詢和存儲的速度也愈來愈慢。

更重要的是一個 Index 實際上是有存儲上限的（除非你設置足夠多的 Shards 和機器），如官方聲明單個 Shard 的文檔數不能超過 20 億（受限於 Lucene index，每一個 Shard 是一個 Lucene index）。

考慮到 I、O，針對 Index 每一個 Node 的 Shards 數最好不超過 3 個，那面對這樣一個龐大的 Index，咱們是採用更多的 Shards，仍是更多的 Index，咱們如何選擇？

Index 的 Shards 總量也不宜太多，更多的 Shards 會帶來更多的 I、O 開銷，其實答案就已經很明確，除非你能接受長時間的查詢等待。

Index 拆分的思路很簡單，時序索引有一個好處就是隻有增長，沒有變動，按時間累積，自然對索引的拆分友好支持，能夠按照時間和數據量作任意時間段的拆分。

ES 提供的 Rollover Api + Index Template 能夠很是便捷和友好的實現 Index 的拆分工做，把單個 index docs 數量控制在百億內，也就是一個 Index 默認 5 個 Shards 左右便可，保證查詢的即時響應。

簡單介紹一下 Rollover API 和 Index Template 這兩個東西，如何實現 index 的拆分。

一、Index Template

咱們知道 ES 能夠爲同一目的或同一類索引建立一個 Index Template，以後建立的索引只要符合匹配規則就會套用這個 Template，沒必要每次指定 Settings 和 Mappings 等屬性。

一個 Index 能夠被多個 Template 匹配，那 Settings 和 Mappings 就是多個 Template 合併後的結果。

有衝突經過 Template 的屬性"order" : 0 從低到高覆蓋（這部分聽說會在 ES6 中會作調整，更好的解決 Template 匹配衝突問題）。

示例：

        PUT _template/template_1{

        "index_patterns" : ["log-*"],

        "order" : 0,

        "settings" : {

            "number_of_shards" : 5

        },

    "aliases" : {

         "alias1" : {}

        }

    }
複製代碼

二、Rollover Index

Rollover Index 能夠將現有的索引經過必定的規則，如數據量和時間，索引的命名必須是 logs-000001 這種格式，並指定 aliases，示例：

    PUT /logs-000001

    {

         "aliases": {

         "logs_write": {}

        }

    }

    # Add > 1000 documents to logs-000001

     POST /logs_write/_rollover

     {

      "conditions": {

            "max_age": "7d",

             "max_docs": 1000

        }

    }
複製代碼

先建立索引並指定別名 logs_write，插入 1000 條數據，而後請求 _rollover api 並指定拆分規則。

若是索引中的數據大於規則中指定的數據量或者時間過期，新的索引將被建立，索引名稱爲 logs-000002，並根據規則套用 Index Template， 同時別名 logs_write 也將被變動到 logs-000002。

注意事項：

索引命名規則必須如同：logs-000001。

索引必須指定 aliases。

Rollover Index API 調用時纔去檢查索引是否超出指定規則，不會自動觸發，須要手動調用，能夠經過 Curator 實現自動化。

若是符合條件會建立新的索引，老索引的數據不會發生變化，若是你已經插入 2000 條，拆分後仍是 2000 條。

插入數據時必定要用別名，不然你可能一直在往一個索引裏追加數據。

技巧是按日期滾動索引：

PUT /<logs-{now/d}-1>

{

"aliases": {

"logs_write": {}

}

}
複製代碼

假如生成的索引名爲 logs-2017.04.13-1，若是你在當天執行 Rollover 會生成 logs-2017.04.13-000001，第二天的話是 logs-2017.04.14-000001。

這樣就會按日期進行切割索引，那若是你想查詢 3 天內的數據能夠經過日期規則來匹配索引名，如：

GET /<logs-{now/d}-*>,<logs-{now/d-1d}-*>,<logs-{now/d-2d}-*>/_search
複製代碼

到此，咱們就能夠經過 Index Template 和 Rollover API 實現對 Index 的任意拆分，並按照須要進行任意時間段的合併查詢，這樣只要你的時間跨度不是很大，查詢速度通常能夠控制在毫秒級，存儲性能也不會遇到瓶頸。

4、Hot-Warm 架構

冷熱架構，爲了保證大規模時序索引實時數據分析的時效性，能夠根據資源配置不一樣將 Data Nodes 進行分類造成分層或分組架構。

一部分支持新數據的讀寫，另外一部分僅支持歷史數據的存儲，存放一些查詢發生機率較低的數據。

即 Hot-Warm 架構，對 CPU，磁盤、內存等硬件資源合理的規劃和利用，達到性能和效率的最大化。

咱們能夠經過 ES 的 Shard Allocation Filtering 來實現 Hot-Warm 的架構。

實現思路以下：

將 Data Node 根據不一樣的資源配比打上標籤，如：Host、Warm。

定義 2 個時序索引的 Index Template，包括 Hot Template 和 Warm Template，Hot Template 能夠多分配一些 Shard 和擁有更好資源的 Hot Node。

用 Hot Template 建立一個 Active Index 名爲 active-logs-1，別名 active-logs，支持索引切割。

插入必定數據後，經過 roller over api 將 active-logs 切割，並將切割前的 Index 移動到 Warm Nodes 上，如 active-logs-1，並阻止寫入。

經過 Shrinking API 收縮索引 active-logs-1 爲 inactive-logs-1，原 Shard 爲 5，適當收縮到 2 或 3，能夠在 Warm Template 中指定，減小檢索的 Shard，使查詢更快。

經過 force-merging api 合併 inactive-logs-1 索引每一個 Shard 的 Segment，節省存儲空間。

刪除 active-logs-1。

一、Hot，Warm Nodes

Hot Nodes

擁有最好資源的 Data Nodes，如更高性能的 CPU、SSD 磁盤、內存等資源，這些特殊的 Nodes 支持索引全部的讀、寫操做。

若是你計劃以 100 億爲單位來切割 Index，那至少須要三個這樣的 Data Nodes，Index 的 Shard 數爲 5，每一個 Shard 支持 20 億 Documents 數據的存儲。

爲這類 Data Nodes 打上標籤，以便咱們在 Template 中識別和指定，啓動參數以下：

./bin/elasticsearch -Enode.attr.box_type=hot

或者配置文件：

node.attr.box_type: hot

Warm Nodes

存儲只讀數據，而且查詢量較少，但用於存儲長多時間歷史數據的 Data Nodes，這類 Nodes 相對 Hot Nodes 較差的硬件配置，根據需求配置稍差的 CPU、機械磁盤和其餘硬件資源，至於數量根據須要保留多長時間的數據來配比，一樣須要打上標籤，方法跟 Hot Nodes 同樣，指定爲 Warm，box_type: warm。

二、Hot，Warm Template

Hot Template

咱們能夠經過指定參數"routing.allocation.include.box_type": "hot"，讓全部符合命名規則索引的 Shard 都將被分配到 Hot Nodes 上：

PUT _template/active-logs

{

"template": "active-logs-*",

"settings": {

"number_of_shards": 5,

"number_of_replicas": 1,

"routing.allocation.include.box_type": "hot",

"routing.allocation.total_shards_per_node": 2

},

"aliases": {

"active-logs": {}

}

}
複製代碼

Warm Template

一樣符合命名規則索引的 Shard 會被分配到 Warm Nodes 上，咱們指定了更少的 Shards 數量和複本數。

注意，這裏的複本數爲 0，和 best_compression 級別的壓縮，方便作遷移等操做，以及進行一些數據的壓縮：

PUT _template/inactive-logs

{

"template": "inactive-logs-*",

"settings": {

"number_of_shards": 1,

"number_of_replicas": 0,

"routing.allocation.include.box_type": "warm",

"codec": "best_compression"

}

}
複製代碼

假如咱們已經建立了索引 active-logs-1 ，固然你能夠經過 _bulk API 快速寫入測試的數據，而後參考上文中介紹的 Rollover API 進行切割。

三、Shrink Index Rollover API 切割之後，active-logs-1 將變成一個冷索引，咱們將它移動到 Warm Nodes 上。

先將索引置爲只讀狀態，拒絕任何寫入操做，而後修改 index.routing.allocation.include.box_type 屬性，ES 會自動移動全部 Shards 到目標 Data Nodes 上：

PUT active-logs-1/_settings

{

"index.blocks.write": true,

"index.routing.allocation.include.box_type": "warm"

}
複製代碼

Cluster Health API 能夠查看遷移狀態，完成後進行收縮操做，其實至關於複製出來一個新的索引，舊的索引還存在。

POST active-logs-1/_shrink/inactive-logs-1

咱們能夠經過 Head 插件查看整個集羣索引的變化狀況。

Forcemerge

到目前爲止咱們已經實現了索引的冷熱分離，和索引的收縮，咱們知道每一個 Shard 下面由多個 Segment 組成，那 inactive-logs-1 的 Shard 數是 1，但 Segment 仍是多個。

這類索引不會在接受寫入操做，爲了節約空間和改善查詢性能，經過 Forcemerge API 將 Segment 適量合併：

PUT inactive-logs-1/_settings

{ "number_of_replicas": 1 }
複製代碼

ES 的 Forcemerge 過程是先建立新的 Segment 刪除舊的，因此舊 Segment 的壓縮方式 best_compression 不會在新的 Segment 中生效，新的 Segment 仍是默認的壓縮方式。

如今 inactive-logs-1 的複本仍是 0，若是有須要的話，能夠分配新的複本數：

PUT inactive-logs-1/_settings

{ "number_of_replicas": 1 }
複製代碼

最後刪除 active-logs-1，由於咱們已經爲它作了一個查詢複本 inactive-logs-1。

DELETE active-logs-1

走到這裏，咱們就已經實現了 Index 的 Hot-Warm 架構，根據業務特色將一段時間的數據放在 Hot Nodes，更多的歷史數據存儲於 Warm Nodes。

5、其餘優化方案

這一部分咱們會展現更多的優化方案。

一、索引隔離

在多條業務線的索引共用一個 ES 集羣時會發生流量被獨吃獨佔的狀況，由於你們都共用相同的集羣資源，流量大的索引會佔用大部分計算資源而流量小的也會被拖慢，得不到即時響應，或者說業務流量大的索引能夠按天拆分，幾個流量小的索引能夠按周或月拆分。

這種狀況下咱們能夠將規模大的索引和其餘相對小規模的索引獨立存儲，分開查詢或合併查詢。

除了 Master Nodes 之外，Data Nodes 和 Coordinating Nodes 均可以獨立使用（其實若是每一個索引的量都特別大也應該採用這種架構），還有一個好處是對方的某個 Node 掛掉，本身不受影響。

一樣利用 ES 支持 Shard Allocation Filtering 功能來實現索引的資源獨立分配，先將 Nodes 進行打標籤，劃分區域，相似於 Hot-Warm 架構：

node.attr.zone=zone_a、node.attr.zone=zone_b
複製代碼

或者：

node.attr.zone =zone_hot_a、node.attr.zone=zone_hot_b
複製代碼

等打標籤的方式來區別對應不一樣的索引，而後在各自的 Index Template 中指定不一樣的 node.attr.zone 便可。

如"index.routing.allocation.include.zone" : "zone_a,zone_hot_a"，或者排除法"index.routing.allocation.exclude.size": "zone_hot_b"分配到 zone_hot_b 之外的全部 Data Nodes 上。

更多用法能夠參考 Hot-Warm 架構，或 shard-allocation-filtering：

二、跨數據中心

若是你的業務遍及全國各地，四海八荒，若是你數據要存儲到多個機房，若是你的 Index 有幾萬個甚至更多（ Index 特別多，集羣龐大會致使 Cluster State 信息量特別大，由於 Cluster State 包含了全部 Shard、Index、Node 等全部相關信息，它存儲在每一個 Node 上，這些數據發生變化都會實時同步到全部 Node 上，當這個數據很大的時候會對集羣的性能形成影響）。

這些狀況下咱們會考慮部署多個 ES Cluster，那咱們將如何解決跨集羣查詢的問題呢？

目前 ES 針對跨集羣操做提供了兩種方案 Tribe Node 和 Cross Cluster Search。

Tribe Node

須要一個獨立的 Node 節點，加入其餘 ES Cluster，用法有點相似於 Coordinating Only Node。

所不一樣的是 Tribe 是針對多個 ES 集羣之間的全部節點，Tribe Node 收到請求廣播到相關集羣中全部節點，將結果合併處理後返回。

表面上看起來 Tribe Node 將多個集羣串連成了一個總體，遇到同名 Index 發生衝突，會選擇其中一個 Index，也能夠指定：

tribe:

on_conflict: prefer_t1

t1:

cluster.name: us-cluster

discovery.zen.ping.multicast.enabled: false

discovery.zen.ping.unicast.hosts: ['usm1','usm2','usm3']

t2:

cluster.name: eu-cluster

discovery.zen.ping.multicast.enabled: false

discovery.zen.ping.unicast.hosts: ['eum1','eum2','eum3']
複製代碼

Cross Cluster Search

Cross Cluster Search 可讓集羣中的任意一個節點聯合查詢其餘集羣中的數據， 經過配置 elasticsearch.yml 或者 API 來啓用這個功能，API 示例：

PUT _cluster/settings

{

"persistent": {

"search": {

"remote": {

"cluster_one": {

"seeds": [

"127.0.0.1:9300"

]

...

}

}

}

}

}
複製代碼

提交之後整個集羣全部節點都會生效，均可以作爲代理去作跨集羣聯合查詢，不過咱們最好仍是經過 Coordinating Only Nodes 去發起請求。

POST /cluster_one:decision,decision/_search

{

"match_all": {}

}
複製代碼

對集羣 cluster_one 和本集羣中名爲 Decision 的索引聯合查詢。

目前這個功能還在測試階段，但將來可能會取代 Tribe Node，之間的最大的差別是 Tribe Node 須要設置獨立的節點，而 Cross Cluster Search 不須要，集羣中的任意一個節點均可以兼任。

好比能夠用咱們的 Coordinating Only Nodes 作爲聯合查詢節點，另外一個優勢是配置是動態的，不須要重啓節點。

實際上能夠理解爲是一個 ES 集羣之間特定的動態代理工具，支持全部操做，包括 Index 的建立和修改，而且經過 Namespace 對 Index 進行隔離，也解決了 Tribe Node 之 Index 名稱衝突的問題。

6、總結

咱們在文中介紹了幾種方案用來解決時序索引的海量數據存儲和查詢的問題，根據業務特色和使用場景來單獨或組合使用可以發揮出意想不到的效果。

特別是 Nodes 之間的讀寫分離、索引拆分、Hot-Warm 等方案的組合應用對索引的查詢和存儲性能有顯著的提高。

另外 Routing 在新版本中增長了 routing_partition_size，解決了 Shard 難以均衡的問題。

若是你的索引 Mapping 中沒有 parent-child 關聯關係能夠考慮使用，對查詢的性能提高很是有效。