搜索引擎 ElasticSearch

ElasticSearch 是什么?

 ElasticSearch 是一款基于 Lucene 开源的 搜索引擎,不但稳定、可靠、快速,同时具备良好的水平扩展能力

主要概念

Cluster 集群

 在一个分布式系统里面,可以通过多个 ElasticSearch 运行实例组成一个集群,这个集群里面有一个节点叫做主节点 (master), ElasticSearch 是去中心化的,所以这里的主节点是动态选举出来的,不存在单点故障
 在同一个子网内,只需要在每个节点上设置相同的集群名, ElasticSearch 就会自动的把这些集群名相同的节点组成一个集群。节点和节点之间通讯,以及节点间的数据分配和平衡,全部都由 ElasticSearch 自动管理
 在外部看来 ElasticSearch 集群就是一个整体

Node 节点

 每一个运行实例称为一个 Node 节点,每一个运行实例既可以在同一机器上,也可以在不同的机器上
 所谓运行实例,就是一个服务器进程,在测试环境中可以在一台服务器上运行多个服务器进程,在生产环境中建议每台服务器运行一个服务器进程

Index 索引

 ElasticSearch 里的索引概念是名词而不是动词,在 ElasticSearch 里它支持多个索引
 优点类似于关系数据库里面每一个服务器可以支持多个数据库是一个道理,在每一索引下面又可以支持多种类型,这又类似于关系数据库里面的一个数据库可以有多张表一样 (暂且这么理解,后面深入了解后,会发现 type 只是 index 里面默认存在的字段,ES 的很多优化和操作都是基于 index 的,并不是基于 type 的)

1
2
Relational DB -> Databases -> Tables -> Rows -> Columns
Elasticsearch -> Indices -> Types -> Documents -> Fields

Shard 分片

 ElasticSearch 它会把一个索引分解为多个小的索引,每一个小的索引就叫做分片
 分片之后就可以把各个分片分配到不同的节点中去

Replica 副本

 ElasticSearch 的每一个分片都可以有 0 到多个副本,而每一个副本也都是分片的完整拷贝。如此设计的好处是,可以用它来增加速度的同时也提高了系统的容错性
 一旦 ElasticSearch 的某个节点数据损坏或则服务不可用的时候,那么就可以用其他节点来代替坏掉的节点,以达到高可用的目标

Recovery 故障恢复

 ElasticSearch 的 Recovery 代表的是数据恢复或者称之为数据重新分布
 当 ElasticSearch 集群中,有节点加入或退出时,它会根据机器的负载对索引分片进行重新分配(另外,当挂掉的节点再次重新启动的时候也会进行数据恢复)

River 数据源

 ElasticSearch River 代表的是一个数据源,这也是其它存储方式 (比如,数据库) 同步数据到 ElasticSearch 的一个方法
 它是以插件方式存在的一个 ElasticSearch 服务,通过读取 River 中的数据并把它索引到 ElasticSearch 当中去,官方支持的 River 类型有 CouchDB、RabbitMQ、Twitter、Wikipedia 等

Gateway 时空门

 Gateway 代表 ElasticSearch 索引的持久化存储方式,ElasticSearch 默认是先把索引存放到内存中去,当内存满了的时候再持久化到硬盘里
 当这个 ElasticSearch 集群关闭或者再次重新启动时,就会从 Gateway 中读取索引数据
 ElasticSearch 支持多种类型的 Gateway,有本地文件系统(默认)、分布式文件系统、HDFS 和 S3 云存储服务 等

discovery.zen 自动节点发现机制

 discovery.zen 代表 ElasticSearch 的自动节点发现机制,而且 ElasticSearch 还是一个基于 P2P 的系统
 首先,集群会以广播的方式去寻找存在的 Node 节点,然后再通过多播协议来进行节点之间的通信,同时也支持点对点的交互操作

Transport 通讯机制(RPC)

 Transport 代表 ElasticSearch 内部的节点或者集群与客户端之间的交互方式
 默认的内部是使用 TCP 协议来进行交互的,同时它支持 HTTP 协议(JSON格式)、Thrift、Servlet、Memcached、ZeroMQ 等多种的传输协议(通过插件方式集成)

安装

下载

 ElasticSearch-2.2.1.tar.gz(ES 5.x 的安装步骤,详见我的另一篇博客《Storm 与 Kafka 的整合之二:Kafka》)

解压并分发

1
2
3
$ tar -zxvf elasticsearch-2.2.1.tar.gz
$ scp -r elasticsearch-2.2.1/ yuzhouwan02:/opt/
$ scp -r elasticsearch-2.2.1/ yuzhouwan03:/opt/

配置

修改 node1 节点的配置

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
$ vim config/elasticsearch.yml
cluster.name: yuzhouwan
node.name: node1
node.master: true
node.data: true
path.data: /data/elastic/data
path.logs: /data/elastic/log
path.work: /data/elastic/work
path.plugins: /data/elastic/plugin
index.number_of_shards: 3
index.number_of_replicas: 1
network.bind_host: 192.168.1.101
network.publish_host: 192.168.1.101
transport.tcp.port: 9300
transport.tcp.compress: true
client.transport.sniff: true
http.port: 9200
discovery.zen.ping.timeout: 3s
discovery.zen.ping.multicast.enabled: true
discovery.zen.ping.unicast.hosts: ["192.168.1.101", "192.168.1.102", "192.168.1.103"]
http.cors.enabled: true
http.cors.allow-origin: /.*/
http.cors.allow-credentials: true
index.version.created: 2020199

修改 node2 节点的配置

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
$ vim config/elasticsearch.yml
cluster.name: yuzhouwan
node.name: node2
node.master: false
node.data: true
path.data: /data/elastic/data
path.logs: /data/elastic/log
path.work: /data/elastic/work
path.plugins: /data/elastic/plugin
index.number_of_shards: 3
index.number_of_replicas: 1
network.bind_host: 192.168.1.102
network.publish_host: 192.168.1.102
transport.tcp.port: 9300
transport.tcp.compress: true
client.transport.sniff: true
http.port: 9200
discovery.zen.ping.timeout: 3s
discovery.zen.ping.multicast.enabled: true
discovery.zen.ping.unicast.hosts: ["192.168.1.101", "192.168.1.102", "192.168.1.103"]
http.cors.enabled: true
http.cors.allow-origin: /.*/
http.cors.allow-credentials: true
index.version.created: 2020199

修改 node3 节点的配置

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
$ vim config/elasticsearch.yml
cluster.name: yuzhouwan
node.name: node3
node.master: false
node.data: true
path.data: /data/elastic/data
path.logs: /data/elastic/log
path.work: /data/elastic/work
path.plugins: /data/elastic/plugin
index.number_of_shards: 3
index.number_of_replicas: 1
network.bind_host: 192.168.1.103
network.publish_host: 192.168.1.103
transport.tcp.port: 9300
transport.tcp.compress: true
client.transport.sniff: true
http.port: 9200
discovery.zen.ping.timeout: 3s
discovery.zen.ping.multicast.enabled: true
discovery.zen.ping.unicast.hosts: ["192.168.1.101", "192.168.1.102", "192.168.1.103"]
http.cors.enabled: true
http.cors.allow-origin: /.*/
http.cors.allow-credentials: true
index.version.created: 2020199

创建 data / log / work / plugin目录

1
2
3
4
5
6
$ mkdir /data/elastic/
$ cd /data/elastic/
$ mkdir data log work plugin
$ scp -r /data/ yuzhouwan02:/
$ scp -r /data/ yuzhouwan03:/

启动

1
2
3
4
5
# -d 使得 ElasticSearch 在后台运行
$ bin/elasticsearch -d
# 在 root 用户下运行 es
$ bin/elasticsearch -Des.insecure.allow.root=true -d

设置 mapping

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
$ curl -XDELETE 'http://192.168.1.101:9200/_template/analyzer_day'
$ curl -XDELETE 'http://192.168.1.101:9200/analyzer/'
$ curl -XPUT 'http://192.168.1.101:9200/_template/analyzer_day' -d '
{
"template": "analyzer_day_*",
"order": 0,
"settings": {
"index": {
"number_of_shards": 3,
"number_of_replicas": 1,
"analysis": {
"analyzer": {
"ignoreCaseAndLikeQuery": {
"type": "simple",
"filter": "lowercase",
"stopwords": ","
}
}
}
}
},
"mappings": {
"analyzer": {
"properties": {
"title": {
"analyzer": "ignoreCaseAndLikeQuery",
"type": "string"
},
"content": {
"type": "string"
},
"date": {
"format": "epoch_millis",
"type": "date"
}
},
"_all": {
"enabled": false
}
}
}
}'

实用技巧

查询数据

匹配所有

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
# 查询所有 index / type 的数据
GET _search
{
"query": {
"match_all": {}
}
}
# 指定一个 index,搜索该 index 下所有 type 的数据
GET /yuzhouwan/_search
# 指定多个 index
GET /yuzhouwan01,yuzhouwan02/_search
# 利用通配符进行指定多个
GET /yuzhouwan*/_search
# 查询指定 index 下指定 type 的数据
GET /yuzhouwan_index/yuzhouwan_type/_search
# 查询指定 index 下多个指定 type 的数据
GET /yuzhouwan_index/yuzhouwan_type1,yuzhouwan_type2/_search
# 查询多个指定 index 下多个指定 type 的数据
GET /yuzhouwan_index1,yuzhouwan_index2/yuzhouwan_type1,yuzhouwan_type2/_search
# 查询所有 index 下多个指定 type 的数据
GET /_all/yuzhouwan_type1,yuzhouwan_type2/_search
{
"query": {
"match_all": {}
}
}

控制展示数量

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# 由于默认只会展示 10 个匹配结果,所以需要设置 `size` 参数来调整
GET /yuzhouwan/_search?size=100
{
"query": {
"match": {
"pattern": "FLUSH"
}
}
}
GET /yuzhouwan/_search
{
"from" : 0, "size" : 100,
"query": {
"match": {
"pattern": "FLUSH"
}
}
}

指定返回字段

1
2
3
4
5
6
7
8
GET /yuzhouwan/_search
{
"_source" : ["message"],
"from" : 0, "size" : 100,
"query": {
"match_all": {}
}
}

范围查询

1
2
3
4
5
6
7
8
9
10
11
GET /yuzhouwan/_search
{
"query": {
"range":{
"CREATETIME": {
"gte": "1496246400000", # 2017-06-01 00:00:00
"lt": "1498838400000" # 2017-07-01 00:00:00
}
}
}
}

聚合查询

Group Count
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
GET /yuzhouwan/_search
{
"size": 0,
"aggs": {
"group_by_link": {
"terms": {
"field": "link",
"size": 100
},
"aggs": {
"sales_bucket_filter": {
"bucket_selector": {
"buckets_path": {
"the_doc_count": "_count"
},
"script": "params.the_doc_count == 2"
}
}
}
}
}
}
# Result:
[{
"key": -1503538282794558000,
"doc_count": 2
},
{
"key": 982015180641152600,
"doc_count": 2
}]
# 【注意】 不可直接用于 `_delete_by_query` 进行删除,因为 aggregation 语句默认自带 `match_all` 查询字段,将会清空所有数据!

过滤查询

字段长度
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
GET /yuzhouwan/_search
{
"query": {
"bool": {
"must": [
{
"range": {
"time": {
"gte": "1514960000000"
}
}
}
],
"filter": {
"script": {
"script": {
"inline": "doc['message'].values.length > 9",
"lang": "painless"
}
}
}
}
}
}
# 【注意】 需要在 mapping 里面设置 message 字段为 keyword 才行

删除数据

1
2
3
4
5
6
7
8
POST yuzhouwan/_delete_by_query
{
"query": {
"match": {
"message": "yuzhouwan.com"
}
}
}

索引

获取已存在的 Mapping 索引格式

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
# localhost 无效的时候,可以尝试用 IP 地址
$ curl -XGET 'host:port/{index}/_mapping/{type}'
$ curl -XGET 'http://192.168.1.101:9200/yuzhouwan_day_20160527/_mapping/'
# 如果仍然无法访问,查看 `config/elasticsearch.yml` 中配置的 http.port 端口是否在监听
$ lsof -i TCP | grep 9200
java 21116 es 9u IPv6 104371174 0t0 TCP *:9200 (LISTEN)
# 如果找不到监听的接口,并且检查 ES 进程是否正常运作
$ lsof -iTCP -sTCP:LISTEN | grep -iE 'es|elasticsearch'
java 21116 es 9u IPv6 104371174 0t0 TCP *:9201 (LISTEN)
java 24777 es 506u IPv6 266659517 0t0 TCP yuzhouwan01.com:vrace (LISTEN)
java 24777 es 700u IPv6 266659638 0t0 TCP yuzhouwan01.com:iua (LISTEN)
node 25161 es 15u IPv4 266679483 0t0 TCP yuzhouwan01.com:9202 (LISTEN)
# 最后验证下 ES 服务是否可用
$ curl -IGET http://localhost:9200
HTTP/1.1 200 OK
Date: Mon, 05 Jun 2017 08:55:44 GMT
Content-Type: text/html
Last-Modified: Fri, 31 Mar 2017 03:07:59 GMT
Accept-Ranges: bytes
Content-Length: 1480
Server: Jetty(9.2.z-SNAPSHOT)
# 获取所有 indices 副本和 types 类别
$ curl -XGET 'http://192.168.1.101:9200/_all/_mapping'
$ curl -XGET 'http://192.168.1.101:9200/_mapping'
# 指定 type 类别
$ curl -XGET 'http://192.168.1.101:9200/_all/_mapping/blog,site'

如何更新索引

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
# 删掉之前的 template 模板
$ curl -XGET 192.168.1.101:9200/_template/template_monitor_log_day
# 删掉之前的 template 模板
$ curl -XDELETE 192.168.1.101:9200/_template/template_monitor_log_day
# 如果想要使现有的 index 索引生效,也需要把之前的 index 数据删除
$ curl -XDELETE 'http://192.168.1.101:9200/monitor_log_day_*/'
# 添加 template 模板
# 如果在粘贴下面大段 PUT 内容的时候,出现 "自动触发 tab 自动补全效果" 的时候,可以 `mkdir temp && cd temp`,在 temp 空文件夹中执行
$ curl -XPUT 192.168.1.101:9200/_template/template_monitor_log_day -d '
{
"template" : "monitor_log_day_*",
"order" : 0,
"settings" : {
"index" : {
"number_of_shards" : 3,
"number_of_replicas" : 1
}
},
"mappings" : {
"access_log":{
"_source": {
"enabled": true
},
"_all": {
"enabled": false
},
"properties": {
"ip": {
"type": "string"
},
"requestLength": {
"type": "integer"
},
"spendMs": {
"type": "long"
},
"attributes": {
"type": "object"
},
"time": {
"type": "date",
"format": "yyyy-MM-dd HH:mm:ss||strict_date_optional_time||epoch_millis"
}
}
}
}
}'

按天滚动新建索引

1
2
3
4
# 开启 auto_create_index 配置,设置符合 `*day*` 正则的索引;并关闭动态 Mapper
$ vim config/elasticsearch.yml
action.auto_create_index: +*day*,-*
index.mapper.dynamic: false

属性转换

1
2
3
4
5
"attributes": {
"webEntry": "ADMIN",
"userId": "admin",
"userName": "admin"
}

TimeZone

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
$ curl -XPOST http://192.168.0.198:9200/cgpboss_month_201609/attack_alarm/_search -d '{
"query" : {
"bool" : {
"must" : [ {
"match" : {
"customerId" : {
"query" : "69",
"type" : "boolean"
}
}
}, {
"range" : {
"startTime" : {
"from" : "2016-09-20 00:00:00,000",
"to" : "2016-09-20 17:47:57,887",
"format" : "yyyy-MM-dd HH:mm:ss,SSS",
"include_lower" : true,
"include_upper" : false
}
}
} ]
}
}
}'

 如果在构建 date 字段相关的查询是,可以直接传入 yyyy-MM-dd HH:mm:ss,SSS 格式的字符串,避免 +08:00 时区带来的困扰。也可以在程序中,将日期中的 local 时区去掉,从而使得日期和 ES 本身的默认时区 +00:00 一致,具体写法如下

1
QueryBuilders.rangeQuery(params.get("timeField")).format("yyyy-MM-dd HH:mm:ss,SSS").gte("2016-09-20 00:00:00,000")

插件

1
2
3
4
$ bin/plugin -install karmi/ElasticSearch-paramedic
# 安装成功后,重启 ElasticSearch 集群
# 访问 http://localhost:9200/_plugin/paramedic/index.html(显示索引的各种信息的统计结果页面)

配置集群

1
$ curl -XGET 'http://localhost:9200/_nodes?os=true&process=true&pretty=true'

技术内幕

ElasticSearch 和 Lucene 的内核是如何实现的,为什么选型 ES?

  • Lucene(基于 Java 的全文检索库)

    需要为了 分布式、高可用性 等特性进行额外的编码、非实时(索引建立 -> 可检索)

  • ElasticSearch(一个实时的分布式搜索和分析引擎)

    分布式、实时分发、实时搜索、易用性(Multi-tenancy 多租户 / Gateway 备份)、高可用性(各节点组成对等的网络结构,方便故障转移)

进一步思考:

1
2
3
4
Solr、Lucene 作为全文检索系统,和 ElasticSearch 这种搜索引擎 有什么本质的区别?
搜索引擎(网页数据的快速采集、海量数据的索引和存储、搜索结果的相关性排序、搜索效率的毫秒级要求、分布式处理和负载均衡、自然语言的理解技术 等等)
全文检索(文本全文检索,包括信息的存储、组织、表现、查询、存取等各个方面,其核心为文本信息的索引和检索)

参考:

为什么 ElasticSearch 集群用 一致性 Hash,而不是 Paxos,这两个算法有什么区别,各自最适的应用场景在哪里?

 一致性 Hash,便于扩容,解决 数据均衡分布的问题
 Paxos 利用最终一致性,规避分布式事务(详见,我的另两篇博客《大数据生态圈的一致性》和《Zookeeper 原理与优化》)

ElasticSearch 索引原理

 已经有太多的书和博客来介绍,这里就不班门弄斧了,就简单画个图来帮助理解 “倒排索引” 的流程吧

Elasticsearch Reverted Index

参考

Blog
原理
信息检索导论学习笔记
Book
  • 《Lucene 原理与代码分析》
  • 《Mastering ElasticSearch》

ElasticSearch 5.0 内核迁移为 Lucene 6.x,是如何利用 Block K-d Tree 来解决 深度分页 问题?

ES 2.x 现状

$(form + size) \, | \, scroll$ 的方式意味着效率低

算法简介

K-d Tree,利用 方差 + 中位数 分割数据入 二叉树,再利用 二叉查找 + 递归溯源 来查找数据

缺憾

N 个节点的 K 维 k-d 树 搜索过程时间复杂度为:Tworst = $O(k*N^{1-1/K})$
所以,无法很好地处理高维数据

参考

整合开发

ElasticSearch + Flume 整合

 https://flume.apache.org/FlumeUserGuide.html

ElasticSearch + IK 中文分词

1
2
3
4
5
6
ElasticSearch 配置中文分词 ik (es 2.3.1 & ik 1.9.1)
http://www.jianshu.com/p/068ee911c001
常见错误:
Unknown Analyzer type [org.ElasticSearch.index.analysis.IkAnalyzerProvider] for [ik]
配置问题,注意 `ik` 和 `es` 的版本,最新版本 "支持的配置项" 会不一致

其他中文分析比对表
ElasticSearch Chinese Word Segmentation

踩过的坑

The number of object passed must be even but was [1]

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
// 建议不使用 JSON.toJSONString,而是自己在 POJO 中拼装 Map<String, Object>
public Map<String, Object> toJSON() {
HashMap<String, Object> json = new HashMap<>();
json.put("systemName", systemName);
json.put("ip", ip);
json.put("path", path);
json.put("time", time);
json.put("message", message);
return json;
}
(ElasticsearchSinkFunction<HBaseServerLog>) (element, ctx, indexer) -> {
_log.debug("Message: {} in ES Sink", element);
if (element == null) return;
indexer.add(indexRequest()
.index(HBASE_SERVER_LOG_INDEX_NAME)
.type(HBASE_SERVER_LOG_TYPE_NAME)
.source(element.toJSON()));
}

参考资料

Doc

Blog

Book

更多资源,欢迎加入,一起交流学习

Technical Discussion Group:(人工智能 1020982(高级)& 1217710(进阶)| BigData 1670647)

  • Post author:Benedict Jin
  • Post link: https://yuzhouwan.com/posts/22654/
  • Copyright Notice: All articles in this blog are licensed under CC BY-NC-SA 4.0 unless stating additionally.