ES使用调研

Posted 2022-01-01 买糖买板栗

1 核心概念

1.1 index 索引

定义：包含一堆有相似结构的文档数据，比如可以有一个客户索引，商品分类索引，订单索引，索引有一个名称。一个index包含很多document，一个index就代表了一类类似的或者相同的document。比如说建立一个product index，商品索引，里面可能就存放了所有的商品数据，所有的商品document。

1.2 Type 类型

每个索引里都可以有一个或多个type，type是index中的一个逻辑数据分类，一个type下的document，都有相同的field，比如博客系统，有一个索引，可以定义用户数据type，博客数据type，评论数据type。

举例：
商品index，里面存放了所有的商品数据，但是商品分很多种类，每个种类的document的field可能不太一样，比如说电器商品，可能还包含一些诸如售后时间范围这样的特殊field；生鲜商品，还包含一些诸如生鲜保质期之类的特殊field。
type，日化商品type，电器商品type，生鲜商品type

• 日化商品type：product_id，product_name，product_desc，category_id，category_name
• 电器商品type：product_id，product_name，product_desc，category_id，category_name，service_period

• 生鲜商品type：product_id，product_name，product_desc，category_id，category_name，eat_period

每一个type里面，都会包含一堆document，document示例：

    "product_id":"2",
    "product_name":"长虹电视机",
    "product_desc":"4k高清",
    "category_id":"3",
    "category_name":"电器",
    "service_period":"1年"



    "product_id":"3",
    "product_name":"基围虾",
    "product_desc":"纯天然，冰岛产",
    "category_id":"4",
    "category_name":"生鲜",
    "eat_period":"7天"

1.3 shard 分片

单台机器无法存储大量数据，es可以将一个索引中的数据切分为多个shard，分布在多台服务器上存储。有了shard就可以横向扩展，存储更多数据，让搜索和分析等操作分布到多台服务器上去执行，提升吞吐量和性能。每个shard都是一个lucene index。

1.4 replica 副本

任何一个服务器随时可能故障或宕机，此时shard可能就会丢失，因此可以为每个shard创建多个replica副本。replica可以在shard故障时提供备用服务，保证数据不丢失，多个replica还可以提升搜索操作的吞吐量和性能。primary shard（建立索引时一次设置，不能修改，默认5个），replica shard（随时修改数量，默认1个），默认每个索引10个shard，5个primary shard，5个replica shard，最小的高可用配置，是2台服务器。

1.5 逻辑

• index包含多个shard
• 每个shard都是一个最小工作单元，承载部分数据，lucene实例，完整的建立索引和处理请求的能力

• 增减节点时，shard会自动在nodes中负载均衡
• primary shard和replica shard，每个document肯定只存在于某一个primary shard以及其对应的replica shard中，不可能存在于多个primary shard

• replica shard是primary shard的副本，负责容错，以及承担读请求负载
• primary shard的数量在创建索引的时候就固定了，replica shard的数量可以随时修改

• primary shard的默认数量是5，replica默认是1，默认有10个shard，5个primary shard，5个replica shard
• primary shard不能和自己的replica shard放在同一个节点上（否则节点宕机，primary shard和副本都丢失，起不到容错的作用），但是可以和其他primary shard的replica shard放在同一个节点上

• primary&replica自动负载均衡，6个shard，3 primary，3 replica
• 每个node有更少的shard，IO/CPU/Memory资源给每个shard分配更多，每个shard性能更好

• 扩容的极限，6个shard（3 primary，3 replica），最多扩容到6台机器，每个shard可以占用单台服务器的所有资源，性能最好
• 超出扩容极限，动态修改replica数量，9个shard（3primary，6 replica），扩容到9台机器，比3台机器时，拥有3倍的读吞吐量

• 3台机器下，9个shard（3 primary，6 replica），资源更少，但是容错性更好，最多容纳2台机器宕机，6个shard只能容纳0台机器宕机
• 这里的这些知识点，你综合起来看，就是说，一方面告诉你扩容的原理，怎么扩容，怎么提升系统整体吞吐量；另一方面要考虑到系统的容错性，怎么保证提高容错性，让尽可能多的服务器宕机，保证数据不丢失

1.6 分片与复制

一个索引可以存储超出单个结点硬件限制的大量数据。比如，一个具有10亿文档的索引占据1TB的磁盘空间，而任一节点都没有这样大的磁盘空间；或者单个节点处理搜索请求，响应太慢。

为了解决这个问题，Elasticsearch提供了将索引划分成多份的能力，这些份就叫做分片。当你创建一个索引的时候，你可以指定你想要的分片的数量。每个分片本身也是一个功能完善并且独立的“索引”，这个“索引”可以被放置到集群中的任何节点上。

分片之所以重要，主要有两方面的原因：

• 允许你水平分割/扩展你的内容容量
• 允许你在分片（潜在地，位于多个节点上）之上进行分布式的、并行的操作，进而提高性能/吞吐量

至于一个分片怎样分布，它的文档怎样聚合回搜索请求，是完全由Elasticsearch管理的，对于作为用户的你来说，这些都是透明的。

在一个网络/云的环境里，失败随时都可能发生，在某个分片/节点不知怎么的就处于离线状态，或者由于任何原因消失了。这种情况下，有一个故障转移机制是非常有用并且是强烈推荐的。为此目的，Elasticsearch允许你创建分片的一份或多份拷贝，这些拷贝叫做复制分片，或者直接叫复制。复制之所以重要，主要有两方面的原因：

• 在分片/节点失败的情况下，提供了高可用性。因为这个原因，注意到复制分片从不与原/主要（original/primary）分片置于同一节点上是非常重要的。
• 扩展你的搜索量/吞吐量，因为搜索可以在所有的复制上并行运行

总之，每个索引可以被分成多个分片。一个索引也可以被复制0次（意思是没有复制）或多次。一旦复制了，每个索引就有了主分片（作为复制源的原来的分片）和复制分片（主分片的拷贝）之别。分片和复制的数量可以在索引创建的时候指定。在索引创建之后，你可以在任何时候动态地改变复制数量，但是不能改变分片的数量。

默认情况下，Elasticsearch中的每个索引被分片5个主分片和1个复制，这意味着，如果你的集群中至少有两个节点，你的索引将会有5个主分片和另外5个复制分片（1个完全拷贝），这样的话每个索引总共就有10个分片。一个索引的多个分片可以存放在集群中的一台主机上，也可以存放在多台主机上，这取决于你的集群机器数量。主分片和复制分片的具体位置是由ES内在的策略所决定的。

1.7 elasticsearch与数据库的类比

2 ES原理

2.1 倒排索引

PUT /megacorp/employee/1  

    "name" :     "John",
    "sex" :      "Male",
    "age" :      25

假设有这么几条数据：

| ID | Name | Age  |  Sex     |
| -- |:------------:| -----:| -----:| 
| 1  | Kate         | 24 | Female
| 2  | John         | 24 | Male
| 3  | Bill         | 29 | Male

ID是Elasticsearch自建的文档id，那么Elasticsearch建立的索引如下:

Name:

| Term | Posting List |
| -- |:----:|
| Kate | 1 |
| John | 2 |
| Bill | 3 |

Age:

| Term | Posting List |
| -- |:----:|
| 24 | [1,2] |
| 29 | 3 |

Sex:

| Term | Posting List |
| -- |:----:|
| Female | 1 |
| Male | [2,3] |

Posting List

Elasticsearch分别为每个field都建立了一个倒排索引，Kate, John, 24, Female这些叫term，而[1,2]就是Posting List。Posting list就是一个int的数组，存储了所有符合某个term的文档id。通过posting list这种索引方式似乎可以很快进行查找，比如要找age=24的同学，爱回答问题的小明马上就举手回答：我知道，id是1，2的同学。但是，如果这里有上千万的记录呢？如果是想通过name来查找呢？

Term Dictionary

Elasticsearch为了能快速找到某个term，将所有的term排个序，二分法查找term，logN的查找效率，就像通过字典查找一样，这就是Term Dictionary。现在再看起来，似乎和传统数据库通过B-Tree的方式类似啊，为什么说比B-Tree的查询快呢？

Term Index

B-Tree通过减少磁盘寻道次数来提高查询性能，Elasticsearch也是采用同样的思路，直接通过内存查找term，不读磁盘，但是如果term太多，term dictionary也会很大，放内存不现实，于是有了Term Index，就像字典里的索引页一样，A开头的有哪些term，分别在哪页，可以理解term index是一颗树：

 这棵树不会包含所有的term，它包含的是term的一些前缀。通过term index可以快速地定位到term dictionary的某个offset，然后从这个位置再往后顺序查找。

 

所以term index不需要存下所有的term，而仅仅是他们的一些前缀与Term Dictionary的block之间的映射关系，再结合FST(Finite State Transducers)的压缩技术，可以使term index缓存到内存中。从term index查到对应的term dictionary的block位置之后，再去磁盘上找term，大大减少了磁盘随机读的次数。

2.2 索引注意事项

将磁盘里的东西尽量搬进内存，减少磁盘随机读取次数(同时也利用磁盘顺序读特性)，结合各种奇技淫巧的压缩算法，用及其苛刻的态度使用内存。所以，对于使用Elasticsearch进行索引时需要注意:

• 不需要索引的字段，一定要明确定义出来，因为默认是自动建索引的
• 同样的道理，对于String类型的字段，不需要analysis的也需要明确定义出来，因为默认也是会analysis的

• 选择有规律的ID很重要，随机性太大的ID(比如java的UUID)不利于查询

2.3 倒排为什么比mysql的b+树索引快？

Mysql 只有 term dictionary 这一层，是以 b-tree 排序的方式存储在磁盘上的。检索一个 term 需要若干次的 random access 的磁盘操作。而 Lucene 在 term dictionary 的基础上添加了 term index 来加速检索，term index 以树的形式缓存在内存中。从 term index 查到对应的 term dictionary 的 block 位置之后，再去磁盘上找 term，大大减少了磁盘的 random access 次数。

 

2.4 分词

表面上我们只要将一段冗长的要检索的目标数据和一串关键字文本丢给它就完事了，事实上ES却不是直接使用完整的关键字文本在完整的目标数据中查找的，它们都要经过一个步骤：拆分成一个个单词、字或词组。如下附录代码中，对document中的属性name进行ik_smart（中文分词器）分词：

// name属性
.startObject("name").field("type", "string")
   // 在文档中存储
   .field("store", "yes")
   // 建立索引
   .field("index", "analyzed")
   // 使用ik_smart进行分词
   .field("analyzer", "ik_smart")......;

ES 的分词不仅仅发生在文档创建的时候，也发生在搜索的时候，如下图所示：

 

3 附录

java使用示例

import java.io.IOException;
import java.net.InetAddress;
import java.util.HashMap;

import org.elasticsearch.action.admin.indices.create.CreateIndexRequestBuilder;
import org.elasticsearch.client.AdminClient;
import org.elasticsearch.client.IndicesAdminClient;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.transport.client.PreBuiltTransportClient;
import org.junit.Before;
import org.junit.Test;

public class AdminAPI 

    private TransportClient client = null;

    // 在所有的测试方法之前执行
    @Before
    public void init() throws Exception 
        // 设置集群名称qxwzlinkx
        Settings settings = Settings.builder().put("cluster.name", "qxwzlinkx")
                // 自动感知的功能(可以通过当前指定的节点获取所有es节点的信息)
                .put("client.transport.sniff", true).build();
        // 创建client
        client = new PreBuiltTransportClient(settings).addTransportAddresses(
                new InetSocketTransportAddress(InetAddress.getByName("192.168.110.1"), 9300),
                new InetSocketTransportAddress(InetAddress.getByName("192.168.110.1"), 9300),
                new InetSocketTransportAddress(InetAddress.getByName("192.168.110.3"), 9300));
    

    /**
     * AdminClient创建索引,并配置一些参数,用来指定一些映射关系等等
     * <p>
     * 这里创建一个索引Index:food,并且指定分区、副本的数量
     */
    @Test
    public void createIndexWithSettings() 
        // 获取Admin的API
        AdminClient admin = client.admin();
        // 使用Admin API对索引进行操作
        IndicesAdminClient indices = admin.indices();
        // 准备创建索引
        indices.prepareCreate("food")
                // 配置索引参数
                .setSettings(
                        // 参数配置器
                        Settings.builder()// 指定索引分区的数量。shards分区
                                .put("index.number_of_shards", 5)
                                // 指定索引副本的数量(注意：不包括本身,如果设置数据存储副本为1,实际上数据存储了2份)
                                // replicas副本
                                .put("index.number_of_replicas", 1))
                // 真正执行
                .get();
    

    /**
     * 你可以通过dynamic设置来控制这一行为,它能够接受以下的选项： true：默认值。
     * <p>
     * 动态添加字段 false：忽略新字段
     * <p>
     * strict：如果碰到陌生字段,抛出异常
     * <p>
     * 给索引添加mapping信息(给表添加schema信息)
     *
     * index:computer
     * type:xiaomi
     *
     * @throws IOException
     */
    @Test
    public void elasticsearchSettingsMappings() throws IOException 
        // 1:settings
        HashMap<String, Object> settings_map = new HashMap<String, Object>(2);
        // shards分区的数量4
        settings_map.put("number_of_shards", 4);
        // 副本的数量1
        settings_map.put("number_of_replicas", 1);

        // 2:mappings(映射、schema)
        // field("dynamic", "true")含义是动态字段
        XContentBuilder builder = XContentFactory.jsonBuilder().startObject().field("dynamic", "true")
                // 设置type中的属性
                .startObject("properties")
                // id属性
                .startObject("id")
                    // 类型是integer
                    .field("type", "integer")
                    // 不分词,但是建索引
                    .field("index", "not_analyzed")
                    // 在文档中存储
                    .field("store", "yes").endObject()
                // name属性
                .startObject("name")
                    // string类型
                    .field("type", "string")
                    // 在文档中存储
                    .field("store", "yes")
                    // 建立索引
                    .field("index", "analyzed")
                    // 使用ik_smart进行分词
                    .field("analyzer", "ik_smart").endObject().endObject().endObject();

        CreateIndexRequestBuilder prepareCreate = client.admin().indices().prepareCreate("computer");
        // 管理索引（user_info）然后关联type（user）
        prepareCreate.setSettings(settings_map).addMapping("xiaomi", builder).get();
    

    /**
     * index这个属性(这个是属性的索引index，不是es的index(库))
     *    no：不建索引
     *    not_analyzed：建索引不分词
     *    analyzed：即分词又建立索引
     *
     *  index:player
     *  type:basketball
     */

    @Test
    public void elasticsearchSettingsPlayerMappings() throws IOException 
        // 1:settings
        HashMap<String, Object> settings_map = new HashMap<String, Object>(2);
        // 分区的数量4
        settings_map.put("number_of_shards", 4);
        // 副本的数量1
        settings_map.put("number_of_replicas", 1);

        // 2:mappings
        XContentBuilder builder = XContentFactory.jsonBuilder().startObject()//
                .field("dynamic", "true").startObject("properties")
                // 在文档中存储、
                .startObject("id").field("type", "integer").field("store", "yes").endObject()
                // 不分词,但是建索引、
                .startObject("name").field("type", "string").field("index", "not_analyzed").endObject()
                //
                .startObject("age").field("type", "integer").endObject()
                //
                .startObject("salary").field("type", "integer").endObject()
                // 不分词,但是建索引、
                .startObject("team").field("type", "string").field("index", "not_analyzed").endObject()
                // 不分词,但是建索引、
                .startObject("position").field("type", "string").field("index", "not_analyzed").endObject()
                // 即分词,又建立索引、
                .startObject("description").field("type", "string").field("store", "no").field("index", "analyzed")
                .field("analyzer", "ik_smart").endObject()
                // 即分词,又建立索引、在文档中存储、
                .startObject("addr").field("type", "string").field("store", "yes").field("index", "analyzed")
                .field("analyzer", "ik_smart").endObject()

                .endObject()

                .endObject();

        CreateIndexRequestBuilder prepareCreate = client.admin().indices().prepareCreate("player");
        prepareCreate.setSettings(settings_map).addMapping("basketball", builder).get();