ElasticSearch详解

一、ElasticSearch是什么

Elasticsearch全称叫全文搜索引擎，简称ES，一个分布式可扩展的实时搜索和分析引擎,一个建立在搜索引擎 Apache Lucene™ 基础上的搜索引擎。Elasticsearch 也是使用 Java 编写的，它的内部使用 Lucene 做索引与搜索，但是它的目的是使全文检索变得简单。
Elasticsearch 不仅仅是 Lucene，并且也不仅仅只是一个全文搜索引擎。 它可以被下面这样准确的形容：
1、一个分布式的实时文档存储，每个字段可以被索引与搜索
2、一个分布式实时分析搜索引擎
3、能胜任上百个服务节点的扩展，并支持 PB 级别的结构化或者非结构化数据

二、ElasticSearch架构

架构图：

ES基本组件说明
1、Client
ES的Client支持http和tcp两种协议去访问，不过目前tcp只支持java。http端口默认9200，tcp端口默认9200,如果是java应用建议是tcp方式访问ES，性能更好。

2、Cluster：集群
ES天生就是集群方式，哪怕只有一个节点。一个集群有一个唯一的名字标志，默认为“elasticsearch”。集群名称非常重要，具体相同集群名的节点才会组成一个集群，集群名称可以在配置文件中指定。

3、EsMaster
EsMaster负责存放Elasticsearch的元数据。 ES元数据包括（身份元数据、索引元数据、文档元数据、路由元数据以及其他类型的元数据），管理集群节点状态。

4、EsNode
Node 节点：存储集群的数据，参与集群的索引和搜索功能。同一个集群内节点的名字不能重复，ES会自动分配，也可以在配置文件中指定。通常在一个节点上分配一个或者多个分片。

5、Shards
分片，就是把索引的数据做水平切分，类似于mysql的分区。在一个多分片的索引中写入数据时，通过路由来确定具体写入那一个分片中，所以在创建索引时需要指定分片的数量，并且分片的数量一旦确定就不能更改。分片还有可以有副本，解决当某个节点宕机后不影响索引正常访问。分片数据量可按照每个分片<30G设置，默认5个分片。

6、Replicas
副本，是指对主分片的备份。主分片和备份分片都可以对外提供查询服务，写操作时先在主分片上完成，然后分发到备份上。当主分片不可用时，会在备份的分片中选举出一个作为主分片，所以备份不仅可以提升系统的高可用性能，还可以提升搜索时的并发性能。但副本数不宜太多，会增加数据写入负担，副本数要<=集群节点数-1。

7、Index（索引-数据库/表）
索引： 一个索引是一个文档的集合。每个索引有唯一的名字，通过这个名字来操作它。Index可以理解为RDBMS里的数据库，也可理解一张表，要看我们怎么使用。

8、Type（类型-表）
类型，指索引内部的逻辑分区，通过Type的名字在索引内进行唯一标识。在查询时如果没有该值，则表示在整个索引中查询。可以理解为表。但是在ES6.x中一个Index只能有一个type，在ES7.x后就取消了type,逐渐减少type是为了提高查询效率。

9、 Document（文档-行）
文档，索引中的每一条数据叫作一个文档。是ES中一个可以被检索的基本单位，每一Document都有一个唯一的ID作为区分，以Json格式来表示。

10、Field（字段-列）
好比关系型数据库中列的概念，一个document有一个或者多个field组成。

11、Mapping
类似于关系型数据库中的表结构信息，用于定义索引中字段（Field）的存储类型、分词方式、是否存储等信息。Elasticsearch中的mapping是可以动态识别的，根据插入的数据自动识别字段类型。一个索引的mapping一旦创建，若已经存储了数据，就不可修改了。还可以创建索引Mapping模板，只要索引名字匹配了就会按照该Mapping插入数据。

12、status 集群状态
指示着当前集群在总体上是否工作正常。它的三种颜色含义如下：
green（绿色）：所有的主分片和副本分片都正常运行。
yellow（黄色）：所有的主分片都正常运行，但不是所有的副本分片都正常运行。
red（红色）：有主分片没能正常运行。

ES与关系型数据库的类比

三、ElasticSearch特点

优点：
1、天生分片，天生集群，从ES出生开始就天然的支持分布式的特征，且无需第三方组件，自带。
2、天生索引，ES 所有数据都是默认进行索引的，这点和mysql正好相反，mysql是默认不加索引，要加索引必须特别说明，ES只有不加索引才需要说明。
3、支持PB级海量数据实时全文搜索。
4、支持多语言访问，支持TCP和RESTful API两种方式访问。

缺点
1、不适合做复杂聚合，会影响ES集群性能。
2、不支持高并发写入数据。
3、ES耗CPU和内存资源，需要用高配置的机器来搭建集群，使用成本比较高。

四、ElasticSearch核心技术分词与倒排索引

1、正向索引
正排表是以文档的 ID 为关键字，表中记录文档中每个 Term 的位置信息，查找时扫描表中每个文档中 Term 的信息直到找出所有包含查询关键字的文档。
因为索引是基于文档建立的，若是有新的文档加入，直接为该文档建立一个新的索引块，挂接在原来索引文件的后面。若是有文档删除，则直接找到该文档号文档对应的索引信息，将其直接删除。但是在查询的时候需对所有的文档进行扫描以确保没有遗漏，数据量较大时，这样就使得检索时间大大延长，检索效率低下。
在我们关系型库中索引为了兼顾插入和查询的性能，都采用了排序树例如：B-Tree/B+Tree这样的数据结构来存储索引。

2、分词
分词就是把字符串按照一定规则分成多个独立的词元（token），Elasticsearch 内置的分词器对中文不友好，会把中文分成单个字来进行全文检索，不能达到想要的结果 。其中IK分词器对中文很好，一般都使用它。

3、倒排索引
倒排索引也可以称反向索引，倒排索引是搜索引擎到核心，主要包括两部分：
(1)单词词典（Term Dictionary）
记录所有文档的单词，一般都比较大
记录单词到倒排列表的关联信息（文档ID）
(2)倒排列表（Posting List）：
记录了单词对应的文档集合，由倒排索引项（Posting）组成
单词词典的实现一般是 B+ Tree

倒排索引项（Posting）主要包含如下信息：
1、文档Id，用于获取原始信息
2、单词频率（TF, Term Frequency），记录该单词在该文档中的出现次数，用于后续相关性算法
3、位置（Position），记录单词在文档中的分词位置（多个），用于做词语搜索（Phrase Query）
4、偏移（Offset），记录单词在文档的开始和结束位置，用于做高亮显示

例子：
原文档（正向索引）

倒排列表（Posting List）

分词和倒排查询时间复杂度都是 O(1)，整个搜索的时间复杂度取决于「求list< id>的交集」，因此搜索实际上问题也变成了求两个集合的交集。

五、ElasticSearch读写过程

1、ES写人数据的过程

1）客户端选择一个node发送请求过去，这个node就是coordinating node（协调节点）
2）coordinating node，对document进行路由，将请求转发给对应的node（有primary shard）
3）实际的node上的primary shard处理请求，然后将数据同步到replica node
4）coordinating node，如果发现primary node和所有replica node都搞定之后，就返回响应结果给客户端.

2、ES读取数据的过程

1）客户端发送请求到任意一个node，成为coordinate node
2）coordinate node对document进行路由，将请求转发到对应的node，此时会使用round-robin随机轮询算法，在primary shard以及其所有replica中随机选择一个，让读请求负载均衡
3）接收请求的node返回document给coordinate node
4）coordinate node返回document给客户端

3、ES数据写入底层原理
（1）数据先写入内存 buffer，在 buffer 里的时候数据是搜索不到的；同时将数据写入 translog 日志文件。如果 buffer 快满了，或者到一定时间，就会将内存 buffer 数据 refresh 到一个新的 segment file 中，但是此时数据不是直接进入 segment file 磁盘文件，而是先进入 os cache 。这个过程就是 refresh。
（2）每隔 1 秒钟，es 将 buffer 中的数据写入一个新的 segment file，每秒钟会产生一个新的磁盘文件 segment file，这个 segment file 中就存储最近 1 秒内 buffer 中写入的数据。但是如果 buffer 里面此时没有数据，那当然不会执行 refresh 操作，如果 buffer 里面有数据，默认 1 秒钟执行一次 refresh 操作，刷入一个新的 segment file 中。
（3）操作系统里面，磁盘文件其实都有一个东西，叫做 os cache，即操作系统缓存，就是说数据写入磁盘文件之前，会先进入 os cache，先进入操作系统级别的一个内存缓存中去。只要 buffer中的数据被 refresh 操作刷入 os cache中，这个数据就可以被搜索到了。
（4）为什么叫 es 是准实时的？ NRT，全称 near real-time。默认是每隔 1 秒 refresh 一次的，所以 es 是准实时的，因为写入的数据 1 秒之后才能被看到。可以通过 es 的 restful api 或者 java api，手动执行一次 refresh 操作，就是手动将 buffer 中的数据刷入 os cache中，让数据立马就可以被搜索到。只要数据被输入 os cache 中，buffer 就会被清空了，因为不需要保留 buffer 了，数据在 translog 里面已经持久化到磁盘去一份了。重复上面的步骤，新的数据不断进入 buffer 和 translog，不断将 buffer 数据写入一个又一个新的 segment file 中去，每次 refresh 完 buffer 清空，translog 保留。随着这个过程推进，translog 会变得越来越大。当 translog 达到一定长度的时候，就会触发 commit 操作。
（5）commit 操作发生第一步，就是将 buffer 中现有数据 refresh 到 os cache 中去，清空 buffer。然后，将一个 commit point写入磁盘文件，里面标识着这个 commit point 对应的所有 segment file，同时强行将 os cache 中目前所有的数据都 fsync 到磁盘文件中去。最后清空 现有 translog 日志文件，重启一个 translog，此时 commit 操作完成。
（6）这个 commit 操作叫做 flush。默认 30 分钟自动执行一次 flush，但如果 translog 过大，也会触发 flush。flush 操作就对应着 commit 的全过程，我们可以通过 es api，手动执行 flush 操作，手动将 os cache 中的数据 fsync 强刷到磁盘上去。
（7）translog 日志文件的作用是什么？你执行 commit 操作之前，数据要么是停留在 buffer 中，要么是停留在 os cache 中，无论是 buffer 还是 os cache 都是内存，一旦这台机器死了，内存中的数据就全丢了。所以需要将数据对应的操作写入一个专门的日志文件 translog 中，一旦此时机器宕机，再次重启的时候，es 会自动读取 translog 日志文件中的数据，恢复到内存 buffer 和 os cache 中去。
（8） translog 其实也是先写入 os cache 的，默认每隔 5 秒刷一次到磁盘中去，所以默认情况下，可能有 5 秒的数据会仅仅停留在 buffer 或者 translog 文件的 os cache 中，如果此时机器挂了，会丢失 5 秒钟的数据。但是这样性能比较好，最多丢 5 秒的数据。也可以将 translog 设置成每次写操作必须是直接 fsync 到磁盘，但是性能会差很多。

其实 es 第一是准实时的，数据写入 1 秒后可以搜索到；可能会丢失数据的。有 5 秒的数据，停留在 buffer、translog os cache、segment file os cache 中，而不在磁盘上，此时如果宕机，会导致 5 秒的数据丢失。

总结一下，数据先写入内存 buffer，然后每隔 1s，将数据 refresh 到 os cache，到了 os cache 数据就能被搜索到（所以我们才说 es 从写入到能被搜索到，中间有 1s 的延迟）。每隔 5s，将数据写入 translog 文件（这样如果机器宕机，内存数据全没，最多会有 5s 的数据丢失），translog 大到一定程度，或者默认每隔 30mins，会触发 commit 操作，将缓冲区的数据都 flush 到 segment file 磁盘文件中。
数据写入 segment file 之后，同时就建立好了倒排索引。

ES删除/更新数据底层原理
（1） 如果是删除操作，commit 的时候会生成一个 .del 文件，里面将某个 doc 标识为 deleted 状态，那么搜索的时候根据 .del 文件就知道这个 doc 是否被删除了。

（2）如果是更新操作，就是将原来的 doc 标识为 deleted 状态，然后新写入一条数据。

（3）buffer 每 refresh 一次，就会产生一个 segment file，所以默认情况下是 1 秒钟一个 segment file，这样下来 segment file 会越来越多，此时会定期执行 merge。每次 merge 的时候，会将多个 segment file 合并成一个，同时这里会将标识为 deleted 的 doc 给物理删除掉，然后将新的 segment file 写入磁盘，这里会写一个 commit point，标识所有新的 segment file，然后打开 segment file 供搜索使用，同时删除旧的 segment file。

六、Java应用使用ElasticSearch

Java应用连接ES建议使用ES的TCP的端口（默认9300），这是ES给java的特权，性能高，其它应用就只能通过http端口（9200）访问了。
springBoot项目建议使用ElasticsearchTemplate
引入jar包：

<!-- ES -->
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-data-elasticsearch</artifactId>
		</dependency>
1
2
3
4
5

在yml配置文件中添加ES连接配置项
spring:
data:
elasticsearch:
cluster-name: my-application
cluster-nodes: 192.168.188.130:9300,192.168.188.53:9300
repositories:
enabled: true

增加ES配置类,不加有时查询会报错

package com.ypc.base.config;

import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;

/**
 * @Author: ypc
 * @Date: 2018-06-30 11:27
 * @Description:
 * @Modified By:
 */
@Configuration
public class ElasticSearchConfig {
    /**
     * 防止netty的bug
     * java.lang.IllegalStateException: availableProcessors is already set to [4], rejecting [4]
     */
    @PostConstruct
    void init() {
        System.setProperty("es.set.netty.runtime.available.processors", "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

这样就可以直接注入ElasticsearchTemplate，对ES进行CRUD

@Autowired
    private ElasticsearchTemplate elasticsearchTemplate;
1
2

springMvc项目不使用ElasticsearchTemplate

引入jar包

		<dependency>
            <groupId>org.elasticsearch.client</groupId>
            <artifactId>transport</artifactId>
            <version>5.1.1</version>
        </dependency>
1
2
3
4
5

XML配置文件里面配置ESUtils

package com.ypc.common.utils;

import com.alibaba.fastjson.JSONObject;
import com.jwd.test.es.DateUtils;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.transport.client.PreBuiltTransportClient;
import org.springframework.beans.factory.InitializingBean;
import java.io.Serializable;
import java.net.InetAddress;

public class ESUtils implements InitializingBean, Serializable {


    //服务器IP
    private static String es_addr;

    //端口号
    private static int es_port;

    //集群名称
    private static String cluster_name;
    private static Settings esSettings;
    /**
     * 嗅探
     */
    private static String client_transport_sniff;
    private static final long serialVersionUID = 1L;
    public static TransportClient client;

    @Override
    public void afterPropertiesSet() throws Exception {
        System.out.println("elasticsearch连接");
        try {
            esSettings = Settings.builder()
                    .put("cluster.name", cluster_name) //设置ES实例的名称
                    .put("client.transport.sniff", client_transport_sniff) //自动嗅探整个集群的状态，把集群中其他ES节点的ip添加到本地的客户端列表中
                    .build();

            client = new PreBuiltTransportClient(esSettings);
            // 创建client
            client.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(es_addr), es_port));
            System.out.println("elasticsearch连接成功");
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("elasticsearch连接失败");
        }

    }


    public  String getEs_addr() {
        return es_addr;
    }

    public  void setEs_addr(String es_addr) {
        this.es_addr = es_addr;
    }

    public  int getEs_port() {
        return es_port;
    }

    public  void setEs_port(int es_port) {
        this.es_port = es_port;
    }

    public  String getCluster_name() {
        return cluster_name;
    }

    public  void setCluster_name(String cluster_name) {
        this.cluster_name = cluster_name;
    }

    public  Settings getEsSettings() {
        return esSettings;
    }

    public  void setEsSettings(Settings esSettings) {
        this.esSettings = esSettings;
    }

    public  String getClient_transport_sniff() {
        return client_transport_sniff;
    }

    public  void setClient_transport_sniff(String client_transport_sniff) {
        this.client_transport_sniff = client_transport_sniff;
    }

    public static void insertData(String userId, String userName, String content) {
        String indexType = "info";
        long operatedTime = com.jwd.common.utils.DateUtils.getCurrentMills();
        JSONObject object = new JSONObject();
        object.put("userId", userId);
        object.put("userName", userName);
        object.put("content", content);
        object.put("operatedTime", operatedTime);
        String dataStr = com.jwd.test.es.DateUtils.getDataStr(operatedTime, "yyyy-MM-dd HH:mm:ss");
        IndexResponse indexResponse = client.prepareIndex(userId, indexType, dataStr).setSource(object).execute().actionGet();
        System.out.println("操作数据已保存，时间：" + indexResponse.getId());
    }

    public static void queryData(String userId, String operatedTime) {
        try {
            QueryBuilder userIdConditon = QueryBuilders.termQuery("userId", userId);

            long startMills = com.jwd.common.utils.DateUtils.startMills(operatedTime);
            long endMills = com.jwd.common.utils.DateUtils.endMills(operatedTime);
            QueryBuilder operatedTimeConditon = QueryBuilders.rangeQuery("operatedTime").from(startMills).to(endMills);
            SearchResponse searchResponse = client.prepareSearch(userId)
                    .setTypes("info")
                    .setQuery(userIdConditon)
                    .setQuery(operatedTimeConditon)
                    .execute()
                    .actionGet();
            SearchHits hits = searchResponse.getHits();
            System.out.println("查到记录数：" + hits.getTotalHits());
            SearchHit[] searchHists = hits.getHits();
            if (searchHists.length > 0) {
                for (SearchHit hit : searchHists) {
                    String uId = (String) hit.getSource().get("userId");
                    String uName = (String) hit.getSource().get("userName");
                    String content = (String) hit.getSource().get("content");
                    long oTime = (long) hit.getSource().get("operatedTime");
                    String dataStr = DateUtils.getDataStr(oTime, "yyyy-MM-dd HH:mm:ss");
                    System.out.println("用户名：" + uName + " 在" + dataStr + " 进行了：" + content);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("查询失败");
        }
    }
}
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

七、ElasticSearch使用注意点

1、ES不适合做实时的复杂聚合，聚合很耗cpu
2、ES不支持高并发写入，可以把写入的数据先放入MQ中，再启消费者批量写入
3、ES集群内存要能存下集群所有数据的一半，才能达到ES最佳状态
4、一个索引数据量不宜太大，大索引影响数据的查询与写入

八、ElasticSearch工具

连接ES的可视化工具
kibana ，可在ES官网下载

elasticsearch-head
node项目可在github下载

git clone git://github.com/mobz/elasticsearch-head.git
cd elasticsearch-head
npm install
1
2
3

启动

npm run start
1

然后浏览器输入如下地址：
http://localhost:9100/

界面如下
主界面

查看所有索引信息

很好用，可以直观看到ES集群信息，可以简单筛选数据，也可自己写ES的DSL语法查询

elasticsearch-dump数据导入导出工具
也是node项目
github下载 https://github.com/taskrabbit/elasticsearch-dump

安装
(local)

npm install elasticdump
./bin/elasticdump
1
2

(global)

npm install elasticdump -g
elasticdump
1
2

从集群把数据导到json文件

elasticdump --input=http://211.159.183.169:9200/aps_job_schedule_index --output=C:/Users/lenovo/Desktop/data/aps_job_schedule_index.json type=data
1

从本地json文件导入ES集群

elasticdump --input=C:/Users/lenovo/Desktop/data/aps_job_schedule_index.json --output=http://211.159.183.169:9200/aps_job_schedule_index type=data
1

从一个ES集群导入另一个ES集群

elasticdump --input=http://211.159.183.169:9200/task_confirm_index --output=http://119.23.235.192:9200/task_confirm_index type=data
1

更多ES信息可查看官网
https://www.elastic.co/cn/elasticsearch/

还有中文说明文档
https://www.elastic.co/guide/cn/elasticsearch/guide/current/index.html

参考链接
https://www.jianshu.com/p/28fb017be7a7/
https://www.wenyuanblog.com/blogs/elasticsearch-forward-index-and-inverted-index.html
https://www.cnblogs.com/756623607-zhang/p/10598043.html