天天看點
傳回

用Elasticsearch做大規模資料的多字段、多類型索引檢索

本文同時釋出在我的個人部落格

之前嘗試了用mysql做大規模資料的檢索優化，可以看到單字段檢索的情況下，是可以通過各種手段做到各種類型索引快速檢索的，那是一種相對簡單的場景。

但是實際應用往往會複雜一些 —— 各類索引（關鍵詞比對、全文檢索、時間範圍）混合使用，還有排序的需求。這種情況下mysql就有點力不從心了，複雜的索引類型，在多索引檢索的時候對每個字段單獨建索引于事無補，而聯合索引無法在如此複雜的索引類型下建起來。

用ElasticSearch來解決這個場景的問題就要簡單的多了。那麼如何用elastic來解決這個問題呢？ 還是帶着業務需求來實踐一遍吧：

①檢索字段有7個，4個關鍵詞比對，1個特殊要求的

a=b&c=d

的分段全文檢索，1個中文全文檢索，1個時間範圍

②資料量很大，需要支援3個月資料的檢索，最好能按月建索引，友善冷備、恢複

1. ElasticSearch Demo Server/Client 環境搭建

為了快速學習elasticsearch的api，可以在本地快速搭建一個demo環境

1.1 elasticsearch server

step1.安裝jdk1.8

https://www.oracle.com/technetwork/java/javase/downloads/index.html 官網下載下傳安裝、配置好環境變量即可

step2.安裝elasticsearch

https://www.elastic.co/cn/downloads/elasticsearch 同樣的，官網下載下傳對應平台的包，這個甚至不需要，直接加壓，就可以在bin目錄下看到服務的啟動檔案

我使用的是windows平台版本的，運作bin目錄下elasticsearch.bat,稍等片刻，通路 http://localhost:9200

用Elasticsearch做大規模資料的多字段、多類型索引檢索

看到此截圖說明elasticsearch demo server啟動成功。

step3.安裝中文分詞器

文章開頭的需求中提到，有需要中文分詞全文索引的字段，是以需要額外安裝一下中文分詞器。

https://github.com/medcl/elasticsearch-analysis-ik/tree/v7.0.0 上官網下載下傳對應elasticsearch版本tag的ik源碼包，比如我使用的最新版本7.0.0，ik也需要下載下傳對應版本的。

elasticsearh是用java寫的，需要安裝maven以編譯此項目。http://maven.apache.org/download.cgi官網下載下傳對應平台的安裝包，編譯或解壓，配置好環境變量。

解壓ik代碼壓縮包，在其根目錄運作

mvn clean && mvn compile && mvn package

，編譯打包

用Elasticsearch做大規模資料的多字段、多類型索引檢索

将target/releasa下生成的編譯好的檔案，解壓到

elasticsearch/plugin/ik

目錄下，重新開機elasticsearch，啟動成功則說明安裝成功（或者直接在github下載下傳對應的release版本）。ik分詞器沒辦法直接測試，需要先建好index，再在index下的分詞器中測試，在後文進行。

1.2 elasticsearch client

elasticsearch server以http協定接口的方式提供服務，官方提供了用戶端的nodejs sdk ：https://github.com/elastic/elasticsearch-js，文檔在這裡https://www.elastic.co/guide/en/elasticsearch/client/javascript-api/current/api-reference.html#_index

用法這裡先不贅述。

2.ElasticSearch使用

正如官網所說，elasticsearch是一個簡單的引擎，同時也是一個複雜的引擎，提供不同級别的配置以實作不同複雜度的需求。網上elastic入門的文章，大多以比較簡單的方式介紹入門級别的使用，但是真正用到産品中的時候，還是要思考一些問題： 如何配置索引字段、如何發起檢索請求、如何添加額外配置。帶着這些問題通讀一遍官網文檔，再來真正使用它，相對來說是比較好一點的。

2.1建立Index

寫入資料之前，首先得考慮如何建立各種不同類型的index，以滿足分詞、檢索、排序、統計的需求，官方文檔對這塊的描述在這裡：https://www.elastic.co/guide/en/elasticsearch/reference/current/mapping.html，真正開始建立index之前，推薦把mapping的文檔通讀一遍，這樣才知道如何選擇合适的type。

2.1.1 Keyword類型字段

首先是4個關鍵詞索引的字段，比較簡單，直接建立keyword類型的property就可以了：

const { Client } = require('@elastic/elasticsearch')
const client = new Client({ node: 'http://localhost:9200' })

client.indices.create({
        index: "asrtest1",
        include_type_name: false,
        body:{
            "mappings" : {
                "properties" : {
                    "id" :       { "type" : "keyword" },
                    "app_key" :  { "type" : "keyword" },
                    "guid" :     { "type" : "keyword" },
                    "one_shot" : { "type" : "keyword" }
                }
            }
        }
    }).then((data)=>{
        console.log("index create success:")
    }).catch((err)=>{
        console.error("index create error:", err)
    })

2.1.2 Date類型字段

然後是1個日期字段，這裡日期的格式是格式化後的

2019-04-15 14:54:01

或者毫秒數，參考mapping文檔中date類型字段的說明，向上面建立的index中插入date類型的字段，并指定字段的格式化方法：

client.indices.putMapping({
        index: "asrtest1",
        include_type_name: false,
        body:{
            "properties" : {
                "log_time" : {
                    "type":   "date",
                    "format": "yyyy-MM-dd HH:mm:ss||epoch_millis"
                }
            }
        }
    }).then((data)=>{
        console.log("add success:", data)
    }).catch((err)=>{
        console.error("add error:", err)
    })

2.1.3 自定義分詞Text類型字段

然後是1個特殊要求的

a=b&c=d

的分段全文檢索， 為了達到這個目标，我們需要使用一個分詞器，僅使用

&

分詞。

這裡需要參考兩處文檔，一處是mapping文檔中text類型字段說明，另一處是整個analysis部分的文檔（描述了分詞器的組成部分、工作原理、如何自定義分詞器等）

一個analyzer由

Character Filters

、

Tokenizers

Token Filters

三部分組成，我們可以自己實作一個自定義分詞器toenizer用于此需求，也可以直接使用内置的pattern analyzer，兩者沒啥差別，這裡圖簡單就用一下内置的Pattern Analyzer：

//1. 關閉asrtest1index
    client.indices.close({index:"asrtest1"}).then(function () {
        //2. 為index添加pattern analyzer
        return client.indices.putSettings({
            index: "asrtest1",
            body:{
                "analysis": {
                    "analyzer": {
                        "qua_analyzer": {
                            "type":      "pattern",
                            "pattern":   "&"
                        }
                    }
                }
            }
        })
    }).then(()=>{
        //3. 為index添加text類型的qua字段，并應用pattern analyzer
        return  client.indices.putMapping({
            index: "asrtest1",
            include_type_name: false,
            body:{
                "properties" : {
                    "qua" : {
                        "type":   "text",
                        "analyzer": "qua_analyzer",
                        "search_analyzer":"qua_analyzer"
                    }
                }
            }
        })
    }).then(()=>{
        //4. 打開asrtest1
        return client.indices.open({index:"asrtest1"})
    }).then(()=>{
        console.log("add qua success")
    }).catch((err)=>{
        console.error("add error:", err.message, err.meta.body)
    })

這裡有個小插曲，直接putSettings的時候，報這個錯，是以上面的代碼中先關閉index，再添加，再open

用Elasticsearch做大規模資料的多字段、多類型索引檢索

至此自定義分詞的qua字段添加完畢，好了，測試一下：

client.indices.analyze({
        "index":"asrtest1",
        "body":{
            "analyzer" : "qua_analyzer",
            "text" : "a=b&c=d&e=f&g=h"
        }
    }).then((data)=>{
        console.log("analyzer run success:", data.body.tokens)
    }).catch((err)=>{
        console.error("analyzer run error:", err)
    })
           
[{token:'a=b',start_offset:0,end_offset:3,type:'word',position:0},
{token:'c=d',start_offset:4,end_offset:7,type:'word',position:1},
{token:'e=f',start_offset:8,end_offset:11,type:'word',position:2},
{token:'g=h',start_offset:12,end_offset:15,type:'word',position:3}]

結果是符合預期的。

2.1.3 中文分詞Text類型字段

elasticsearch通過插件的形式來支援中文分詞，在1.1小節中，我們已經安裝了elasticsearch提供的中文分詞插件，現在來應用并測試一下。

添加text中文分詞字段：

client.indices.putMapping({
        index: "asrtest1",
        include_type_name: false,
        body:{
            "properties" : {
                "text" : {
                    "type":   "text",
                    "analyzer": "ik_max_word",
                    "search_analyzer":"ik_max_word"
                }
            }
        }
    }).then((data)=>{
        console.log("add success:", data)
    }).catch((err)=>{
        console.error("add error:", err)
    })

測試中文分詞引擎：

client.indices.analyze({
        "index":"asrtest1",
        "body":{
            "analyzer" : "ik_max_word",
            "text" : "中韓漁警沖突調查：韓警平均每天扣1艘中國漁船"
        }
    }).then((data)=>{
        console.log("analyzer run success:", data.body.tokens)
    }).catch((err)=>{
        console.error("analyzer run error:", err)
    })
           
[{token:'中韓',start_offset:0,end_offset:2,type:'CN_WORD',position:0},
{token:'漁',start_offset:2,end_offset:3,type:'CN_CHAR',position:1},
{token:'警',start_offset:3,end_offset:4,type:'CN_CHAR',position:2},
{token:'沖突',start_offset:4,end_offset:6,type:'CN_WORD',position:3},
{token:'調查',start_offset:6,end_offset:8,type:'CN_WORD',position:4},
{token:'韓',start_offset:9,end_offset:10,type:'CN_CHAR',position:5},
{token:'警',start_offset:10,end_offset:11,type:'CN_CHAR',position:6},
{token:'平均',start_offset:11,end_offset:13,type:'CN_WORD',position:7},
{token:'每天',start_offset:13,end_offset:15,type:'CN_WORD',position:8},
{token:'扣',start_offset:15,end_offset:16,type:'CN_CHAR',position:9},
{token:'1',start_offset:16,end_offset:17,type:'ARABIC',position:10},
{token:'艘',start_offset:17,end_offset:18,type:'COUNT',position:11},
{token:'中國',start_offset:18,end_offset:20,type:'CN_WORD',position:12},
{token:'漁船',start_offset:20,end_offset:22,type:'CN_WORD',position:13}]

2.2 按月自動建索引

2.1小節中分步驟分析了每個字段應該如何建立索引，而業務場景下有個需求是按月建索引。可以選擇跑個定時腳本，每個月去自動建立下一個月的index，也有更簡單的選擇 —— 插入資料的時候，如果發現索引名稱不存在，則自動建立索引，elasticserch提供了這樣的功能。為了實作這個目标，需要看兩個部分的文檔： 叢集的自動index建立配置、index模闆。

首先建立一個index模闆：

client.indices.putTemplate({
        "name": "asrtemp",
        "include_type_name": false,
        "body":{
            "index_patterns" : ["asr*"],
            "settings": {
                "analysis": {
                    "analyzer": {
                        "qua_analyzer": {
                            "type":      "pattern",
                            "pattern":   "&"
                        }
                    }
                }
            },
            "mappings": {
                "properties": {
                    "id" :       { "type" : "keyword" },
                    "app_key" :  { "type" : "keyword" },
                    "guid" :     { "type" : "keyword" },
                    "one_shot" : { "type" : "keyword" },
                    "log_time" : {
                        "type":   "date",
                        "format": "yyyy-MM-dd HH:mm:ss||epoch_millis"
                    },
                    "qua" : {
                        "type":   "text",
                        "analyzer": "qua_analyzer",
                        "search_analyzer":"qua_analyzer"
                    },
                    "text" : {
                        "type":   "text",
                        "analyzer": "ik_max_word",
                        "search_analyzer":"ik_max_word"
                    }
                }
            }
        }
    }).then((data)=>{
        console.log("add template success:", data)
    }).catch((err)=>{
        console.error("add template error:", err)
    })

修改叢集配置，設定自動建立索引時隻應用此模闆（也可以修改此配置，預設是應用所有滿足pattern的模闆）：

client.cluster.putSettings({
        body:{
            "persistent": {
                "action.auto_create_index": "asrtemp"
            }
        }
    })

寫入一個document，指定一個不存在但是滿足template中index_patterns的index：

client.create({
        id:"testid1",
        index:"asrtest2",
        body:{
            "id":"testid1",
            "app_key":"dc6aca3e-bc9f-45ae-afa5-39cc2ca49158",
            "guid":"4ccaee22-5ce3-11e9-9191-1b9bd38b79e0",
            "one_shot":"0",
            "log_time":"2019-05-14 09:13:20",
            "qua":"key1=asd.asda.asf&key2=2.0.20.1&key3=val3&testk=testv",
            "text":"明天武漢的天氣好不好啊"

        }
    }).then((data)=>{
        console.log("create success:", data)
    }).catch((err)=>{
        console.error("create error:", err)
    })

寫入成功，查詢一下新生成的index的資訊：

client.indices.get({
        index: "asrtest2"
    }).then((data)=>{
        console.log("get success:", data.body.asrtest2.mappings,data.body.asrtest2.settings)
    }).catch((err)=>{
        console.error("get error:", err)
    })
           
{ properties: 
   { app_key: { type: 'keyword' },
     guid: { type: 'keyword' },
     id: { type: 'keyword' },
     log_time: { type: 'date', format: 'yyyy-MM-dd HH:mm:ss||epoch_millis' },
     one_shot: { type: 'keyword' },
     qua: { type: 'text', analyzer: 'qua_analyzer' },
     text: { type: 'text', analyzer: 'ik_max_word' } } } 
{ index: 
   { number_of_shards: '1',
     provided_name: 'asrtest2',
     creation_date: '1555319547412',
     analysis: { analyzer: [Object] },
     number_of_replicas: '1',
     uuid: '_GcwsE4vSBCDW0Pv35w0uA',
     version: { created: '7000099' } } }

與模闆是一緻的，自動建立成功。

2.3檢索請求

請求時，需要做到：①各種字段交叉組合檢索 ②支援分頁統計count、offset ③支援按時間排序 ④延時不能太長。下面首先插入幾百萬條模拟資料，然後實踐一下上面的三個檢索需求。

2.3.1 模拟資料

為了快速大批量插入資料，應該使用elasticsearch提供的bulk api來進行資料插入的操作，關鍵代碼：

function mockIndex(index){
    var indexName = "asrtest2"
    if(index >= 4000000){
        console.log(`index mock done!`)
        return;
    }
    if(index % 10000 == 0){
        console.log(`current num:${index}`)
    }
    var mockDataList = []
    for(var i=0;i<500;i++){
        var mock = getOneRandomData(index++)
        mockDataList.push({ "index" : { "_index" : indexName, "_id" : mock.id } })
        //mock: {app_key,log_time,guid,qua,id, text, one_shot"}
        mockDataList.push(mock)
    }
    client.bulk({
        index:indexName,
        body: mockDataList
    }).then(()=>{
        mockIndex(index)
    }).catch((err)=>{
        console.error("bulk error:",err.message)
    })
}

mockIndex(0)

2.3.2 條件檢索

模拟資料灌滿後，測試一下多索引聯合檢索，并設定排序字段、擷取count、觀測性能、驗證結果的正确性。此處建議通讀elasticsearch文檔Search APIs、 Query DSL（elastic自己造的一種抽象文法樹）。

用Elasticsearch做大規模資料的多字段、多類型索引檢索

在這些部分可以看到，各個類型的字段應該如何生成檢索條件

用Elasticsearch做大規模資料的多字段、多類型索引檢索

這些部分可以看到如何将各個字段的檢索條件合理組成一個複合檢索參數

用Elasticsearch做大規模資料的多字段、多類型索引檢索

這些部分可以看到如何使用字段排序、如何設定傳回結果數量、偏移量。

最後，根據文檔的說明，寫一個測試檢索代碼：

client.search({
    index: "asrtest1",
    from: 0,
    size: 10,
    sort: "log_time:asc",
    body:{
        "query": {
            "bool": {
                //測試query和filter
                //"must": [
                //    { "match_phrase": { "qua":   "SDK=39.60.17160906"}},
                //    { "match_phrase": { "text": "舞麟" }}
                //],
                "filter": [

                    { "match_phrase": { "qua":   "SDK=39.60.17160906"}},
                    { "match_phrase": { "text": "舞麟" }},
                    //id和guid是unique字段
                    //{ "term":  { "id": "c2e86a41-5f73-11e9-b3d0-45c4efcbf90f" }},
                    //{ "term":  { "guid": "2081359c-5f72-11e9-b3d0-45c4efcbf90f" }},
                    { "term":  { "app_key": "faf1e695-9a97-4e8f-9339-bdce91d4848a" }},
                    { "term":  { "one_shot": "1" }},
                    { "range": { "log_time": { "gte": "2019-04-15 08:00:00" }}}
                ]
            }
        }
    }
}).then((data)=>{
    console.log("timecost:", data.body.took)
    console.log("total:", data.body.hits.total)
    console.log("hits:", data.body.hits.hits)
}).catch((err)=>{
    console.log("error:", err)
})

變換各種條件查詢，條件查詢、排序、傳回條數、偏移、總數等都是符合預期的。

如果有細心看上面的代碼，可以發現query條件中有注釋掉的

must

部分，這是因為我面臨的業務場景下不需要對document進行score計算，隻需要過濾結果，是以将所有的條件塞到filter中，elastic内部會有一些緩存政策，提高效率。經測試，将兩個

match_phase

條件放到must中，400W條資料檢索平均耗時在30—40ms，而放到filter中後，平均僅為7—8ms。

減少條件，隻保留時間限制，發現：

用Elasticsearch做大規模資料的多字段、多類型索引檢索

至少應該有兩百多萬條結果，這裡total隻有10000條。

可以通過修改index的settings，index.max_result_window屬性，來修改這個數量。

但是！文檔中提到" Search requests take heap memory and time proportional to from + size and this limits that memory"，還有這篇文檔，這裡可以看到es就不适合用于大規模資料的完全周遊！想要使用es完美解決所有問題，得一口老血噴在螢幕上！

這裡雖然沒辦法直接查詢到大offset資料，但是可以通過Count API查詢到真實總數，然後通過其它的search方法來達到分頁的目标，好在elasticsearch也是考慮了這一點，提供了Search After API來應對這種場景。

說白了就是使用了另一種分頁模式，需要業務自己維護上下文，通過傳入上一次查詢的最後一個結果作為起點，再往後面去查詢結果。

修改查詢代碼：

client.search({
    index: "asrtest1",
    size: 10,
    sort: "log_time:desc,id:desc",
    body:{
        "query": {
            "bool": {
                "filter": [
                    { "range": { "log_time": { "gte": "2019-04-15 08:20:00" }}}
                ]
            }
        }
    }
}).then((data)=>{
    console.log("first 10 result:", data.body.hits.hits)
    let last = data.body.hits.hits.slice(-1)[0].sort
    return client.search({
        index: "asrtest1",
        size: 10,
        sort: "log_time:desc,id:desc",
        body:{
            "query": {
                "bool": {
                    "filter": [
                        { "range": { "log_time": { "gte": "2019-04-15 08:20:00" }}}
                    ]
                }
            },
            "search_after": last,
        }
    })
}).then((data)=>{
    console.log("second 10 result:", data.body.hits.hits)
}).catch((err)=>{
    console.log("error:", err)
})

這樣，就可以查到任意多的結果，又不會把叢集搞死了。

本文基于一個簡單的業務場景大緻實踐了一遍elasticsearch的使用，而實際上叢集的搭建、運維，是一個非常複雜的工作，而很多雲服務上都提供了包裝好的PAAS服務，如騰訊雲ElasticSearch Service，直接購買接入即可。

資料檢索 elasticsearch
上一篇: Blazor 初體驗DEMO
下一篇: .net core IdentityServer4 使用query參數

繼續閱讀