ElasticSearch

• 开源的高扩展的分布式全文检索引擎。搜索引擎原理：建立反向索引（倒排索引）；es中的索引、类型（已经抛弃的概念）和文档的概念类似于MySQL中的数据库、表和行。
• 倒排索引
  1. 通过关键字索引具体内容的方式就叫做倒排索引。正排索引：根据标题等方式来索引内容。诸如，通过静夜思来索引“窗前明月光，疑是地上霜。”是正排索引。而通过“前”建立了含有“前”的诗句的索引，通过该关键字的索引来查找具体内容，就是倒排索引。
  2. 为了提升存储效率，采用关键字索引具体诗的id；同时关键字对应多个诗，为此关键字对应一个索引矩阵；也就是多个关键字对应多个诗的id，而诗id可以对应到具体的诗内容；
  3. 搜索引擎的三个步骤：爬取内容、内容分词、建立倒排索引。
• Lucene
  1. 一套用于信息检索的工具包；包括了索引结构、读写索引的工具、相关性工具及排序等功能；elasticsearch是基于它封装的库，elasticsearch在lucene的基础上，还提供了数据获取、解析、分词等功能；

基本操作及原理

• es-head 查看集群
  • 集群健康值：
    1. green：所有的主分片以及副本分片都是active，集群健康
    2. yellow：至少一个副本分片不可用，但是所有主分片是可用的，数据是完整的
    3. red：至少有一个主分片不可用，数据不完整，集群式不可用的
• kibana操作
  • 索引、分片与副本：（索引是我们创建的索引，分片是elasticsearch为我们创建的索引分片存储，而副本是分片在不同于分片所在节点的节点上的存储备份；
    
  • 添加中文分词扩展：elasticsearch-ik（一定要对应elasticsearch版本）
  • 创建索引：

    1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23

    PUT /poems: // 索引 { "settings":{ "number_of_shards" : 3, // 三个分片 "number_of_replicas" : 1, // 每个分片有一个副本 } "mappings": { // 映射，相当于mysql中列的类型和约束 "properties" :{ "author":{ // 一个author列 "type" : "keyword" // 标识不会被分词 }, "name":{ // 一个名字列 "type" : "keyword" }, "content":{ // 一个内容列 "type" : "text", // 可被分词 "analyzer" : "ik_max_word" // 分词策略 // ik_max_word 以最小粒度拆分，越多越好，拆分结果：窗前 明月光 明月 月光 // ik_smart ， 以最大粒度拆分，拆分结果：窗前 明月光 } } } }

  • 分片及副本分布结果：（我们可以发现分片被标号为0、1、2，她们的副本分布结果可以发现：即使任意一个node宕机，我们仍然还可以得到完整的0号、1号和2号分片）
    
  • 数据操作:

    1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76

    /* 插入操作 */ PUT /poems/_doc/1 // 使用put方法插入文档，需要手动指定文档id { "author" : "李白", "name" : "静夜思", "content": "床前明月光，疑是地上霜。举头望明月，低头思故乡。" } POST /poems/_doc // 使用post方法插入文档，会随机生成文档id { "author" : "李白", "name" : "望庐山瀑布", "content": "日照香炉生紫烟，遥看瀑布挂前川。飞流直下三千尺，疑是银河落九天。" } /* 简单查询操作 */ GET /poems/_doc/1 // 根据文档id查询 GET /poems/_analyze // 查看分词结果 { "field" : "content", "text" : [ "窗前明月光，疑是地上霜，举头望明月，低头思故乡" ] } GET /poems/_search GET /poems/_search // 带条件，es不支持join联表查询 { "query" : { // 相当于where "bool" : { "must" : [ // 相当于and { "match_all": {} } ], "must_not" : [], // 相当于not "should" :[] // 相当于or } }, "from":0, "size":10, // from+size == limit 0,9 "sort":[], // order by "aggs":{} // group by } // term 查询，关键字keyword查询 GET /poems/_search { "query":{ "term":{ "content":"挂川" // 不会进行分词，所以查不到 } } } // match 查询，分词text查询 GET /poems/_search { "query":{ "match":{ "content":"挂川" // 会被分解成两个字：挂 川，这样就能查询到 } } } /* 更新操作 */ // 1. PUT 全量更新，id存在，将原来的删除，再添加，所以必须指定全部的字段 PUT /poems/_doc/1 { "author" : "唐-李白", "name" : "静夜思", "content": "床前明月光，疑是地上霜。举头望明月，低头思故乡。" } // 2. POST 增量更新，可以只写需要更新的字段及值 POST /poems/_doc/1/_update { "doc" : { "author" : "唐-李白", } } /* 删除操作 */ DELETE /poems/_doc/1

  • 数据写入流程
    1. 假设client连接的是node0（共3个node），我们需要增上改查的文档id为_id，那么先会对请求进行路由（hash(_id)%3）找到实际需要操作的node（主分片所在位置）；
    2. 主分片处理请求，然后再将修改同步到副本分片上；
    3. 副本分片处理完成后，返回主分片，然后再返回到node0，最后再返回结果给client；
    4. 我们可以进行es的配置，使主分片处理完成后不等副本分片处理就返回，但是不建议这么做；
  • 写入原理：
    1. 每一个在分片中的数据会被分成多个segment缓存，同时会将写操作写入translog；分片节点不断将segment合并，成为大的segment；直到translog足够大的时候，才会将segment落盘，再将translog清空；
    2. 我们的查询操作在写入到segment的缓存后就可以进行了，并不需要等到实际落盘，所以读写效率高；
    3. translog是顺序写，所以就算宕机了，仍可以从translog中进行恢复；
  • 数据查询流程：
    1. 不需要找到主分片，只要有所需要的分片或者副本就可以返回结果了；
  • 更多命令学习：elastic.co/en/guide/elasticsearch/reference/current/index.html

ElasticSearch查询

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

GET /poems/_search
{
    "_source":["name", "author"], // select name, author from table;
    "query":{
        "bool":{
            "must":[
                "term" : {
                    "author" : "李白" // where author=李白
                }
            ],
            "must_not" :[
                "term":{
                    "name" :{  // 和上面那个只是写法上的区别，都可以的
                        "value": "静夜思" // where not name=静夜思
                    }
                }
            ]
        }
    }
}

GET /poems/_search
{
    "query":{
        "bool":{
            "should":[ // where author=李白 or where author=杜甫
                "term" : {
                    "author" : "李白" 
                },
                "term" : {
                    "author" : "杜甫" 
                }
            ],
            "must_not" :[
                "term":{
                    "name" :{  
                        "value": "静夜思" 
                    }
                }
            ]
        }
    }
}

GET /poems/_search
{
    "_source":["name", "author"],
    "query" : {
        "bool" : {
            "filter" : [
                {
                    "range":{
                        "favourite":{
                            "lt":90  // where favourite < 90
                        }
                    }
                }
            ]
        }
    }
}

• 查询过程：用户->倒排索引->文档id->文档->用户
• 有时候文档内容比较多，或并不需要建立倒排索引，指定某些字段不存储在es中：

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21

  PUT /poems: { "settings":{ "number_of_shards" : 3, "number_of_replicas" : 1, }, "mappings": { "properties" :{ "author":{ "type" : "keyword" }, "name":{ "type" : "keyword" }, "content":{ "type" : "text", } }, _source:["content"] //指定content不存储在es中 } }

• ES常见使用方法：
  • es是非关系型“数据库”，并且不支持事务；如何保持数据安全型：
    1. server先访问mysql
    2. mysql同步数据到ElasticSearch（通过canal或go-mysql-tranfer中间件）
    3. 中间件原理：伪装成mysql的从数据库，将mysql的增删改操作同步到es当中
  • 那么，于是不需要倒排索引的内容存储在mysql中，根据es操作查询获取的文档_id去mysql中查询

全文查找

• 前提：存储了全文数据，并建立了倒排索引

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32

  GET /poems/_search { "query" : { "match":{ "content": "到底有多少个" // 先分词再查询 } } } GET /poems/_analyze // 查看如何分词 { "field": "content", "text":[ "到底有多少个" ] } GET /poems/_search { "query" : { "match_phrase":{ // 先分词再查询，需要匹配所有分词，且保证分词的出现顺序不变 "content": "到底有多少个" } } } GET /poems/_search { "query" : { "multi_match":{ // 先分词匹配多个text "query": "到底有多少个", "fields": ["subject", "content"] } } }

分组聚合操作

• 按作者分组并按照favourite组内求和排序：select sum(favourite) from poems group by auther;

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22

  GET /poems/_search { "size" : 0, // 仅查看分组结果，不显示原先命中的内容 "aggs":{ "aggs_author":{ "terms":{ "field":"author", "size":10, "order":{ "sum_fav":"asc" } }, "aggs":{ "sum_fav":{ "sum":{ "field":"favourite" } } } } } }

Golang操作ElasticSearch

• 安装依赖

  1 2	go get github.com/olivere/elastic/v7 # 第三方实现，更新会慢一些，但是使用容易 go get github.com/elastic/go-elasticsearch/v7 # 官方实现

• 基本操作

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180

  package main import ( "context" "encoding/json" "fmt" "log" "github.com/olivere/elastic/v7" ) type poem struct { Id string `json:"id"` Author string `json:"author"` Name string `json:"name"` Content string `json:"content"` Favourite int `json:"favourite"` } func main() { var url = []string{"http://127.0.0.1:9200"} // 集群需要设置多个 es, err := elastic.NewClient(elastic.SetURL(url...)) if err != nil { log.Fatal(err) } /* 打印版本 */ esversion, err := es.ElasticsearchVersion(url[0]) if err != nil { log.Fatal(err) } fmt.Println("es version: ", esversion) /* 操作索引 */ if false { exist, err := es.IndexExists("poems1").Do(context.Background()) if err != nil { log.Fatal(err) } if exist { fmt.Println("poems1 exists!") es.DeleteIndex("poems1").Do(context.Background()) fmt.Println("poems1 delete!") } else { fmt.Println("poems1 not exists!") schema := ` { "settings": { "number_of_shards": 3, "number_of_replicas": 1 }, "mappings": { "properties": { "author": { "type": "keyword" }, "name": { "type": "keyword" }, "content": { "type": "text", "analyzer": "ik_max_word" }, "favourite": { "type": "integer" } } } } ` res, err := es.CreateIndex("poems1").Body(schema).Do(context.Background()) if err != nil { log.Fatal(err) } fmt.Println("result: ", res.Acknowledged) } } /* 查询操作 */ if false { // term查询 termq := elastic.NewTermQuery("author", "李白") res, err := es.Search().Index("poems1").Query(termq).Do(context.Background()) if err != nil { fmt.Println("Err") log.Fatal(err) } fmt.Println(res.TotalHits()) if res.TotalHits() > 0 { for _, hit := range res.Hits.Hits { var p poem err = json.Unmarshal(hit.Source, &p) if err != nil { log.Fatal(err) } p.Id = hit.Id fmt.Printf("%#v\n", p) } } } if false { // match查询 mq := elastic.NewMatchQuery("content", "花落多少") res, err := es.Search().Index("poems1").Query(mq).Do(context.Background()) if err != nil { fmt.Println("Err") log.Fatal(err) } fmt.Println(res.TotalHits()) if res.TotalHits() > 0 { for _, hit := range res.Hits.Hits { var p poem err = json.Unmarshal(hit.Source, &p) if err != nil { log.Fatal(err) } p.Id = hit.Id fmt.Printf("%#v\n", p) } } } if false { // match-phrase查询 mq := elastic.NewMatchPhraseQuery("content", "花落多少") res, err := es.Search().Index("poems1").Query(mq).Do(context.Background()) if err != nil { fmt.Println("Err") log.Fatal(err) } fmt.Println(res.TotalHits()) if res.TotalHits() > 0 { for _, hit := range res.Hits.Hits { var p poem err = json.Unmarshal(hit.Source, &p) if err != nil { log.Fatal(err) } p.Id = hit.Id fmt.Printf("%#v\n", p) } } } if false { // range rq := elastic.NewRangeQuery("favourite").Gt(89) res, err := es.Search().Index("poems1").Query(rq).Do(context.Background()) if err != nil { fmt.Println("Err") log.Fatal(err) } fmt.Println(res.TotalHits()) if res.TotalHits() > 0 { for _, hit := range res.Hits.Hits { var p poem err = json.Unmarshal(hit.Source, &p) if err != nil { log.Fatal(err) } p.Id = hit.Id fmt.Printf("%#v\n", p) } } } if true { // 聚合操作 aq := elastic.NewTermsAggregation().Field("author").Size(10).Order("sum_fav", true) sq := elastic.NewSumAggregation().Field("favourite") aq.SubAggregation("sum_fav", sq) res, err := es.Search().Index("poems1").Aggregation("aggs_author", aq).Size(0).Do(context.Background()) if err != nil { fmt.Println("Err") log.Fatal(err) } agg_author, found := res.Aggregations.Terms("aggs_author") if !found { fmt.Println("cant found result") return } for _, bucket := range agg_author.Buckets { key := bucket.Key sum_fav, found := bucket.Aggregations.Sum("sum_fav") if found { fmt.Println("key:", key, "sum:", *sum_fav.Value) } } } }