[大数据]elasticsearch常用综述

快速入门

1. 基础概念-快速入门

Elasticsearch是一个基于Lucene的搜索服务器。它提供了一个分布式多用户能力的全文搜索引擎，基于RESTful web接口。Elasticsearch是用Java语言开发的，并作为Apache许可条款下的开放源码发布，是一种流行的企业级搜索引擎。Elasticsearch用于云计算中，能够达到实时搜索，稳定，可靠，快速，安装使用方便。官方客户端在Java、.NET（C#）、PHP、Python、Apache Groovy、Ruby和许多其他语言中都是可用的。根据DB-Engines的排名显示，Elasticsearch是最受欢迎的企业搜索引擎，其次是Apache Solr，也是基于Lucene。 – 百度百科

2. 节点 Node、集群 Cluster 和分片 Shards

ElasticSearch 是分布式数据库，允许多台服务器协同工作，每台服务器可以运行多个实例。

单个实例称为一个节点（node），一组节点构成一个集群（cluster）。

分片是底层的工作单元，文档保存在分片内，分片又被分配到集群内的各个节点里，每个分片仅保存全部数据的一部分。

3. 索引 Index、类型 Type 和文档 Document

• Index 对应 MySQL 中的 Database；
• Type 对应 MySQL 中的 Table；
• Document 对应 MySQL 中表的记录。
• Field 对应 MySQL 中的字段

在 7.0 以及之后的版本中 Type 被废弃了。（其实还可以用，但是已经不推荐了）

一个MySQL实例中可以创建多个 Database，一个Database中可以创建多个Table。

ES 的Type 被废弃后：

• ES 实例：对应 MySQL 实例中的一个 Database。
• Index 对应 MySQL 中的 Table 。
• Document 对应 MySQL 中表的记录。
• Field 对应 MySQL 中的字段

准备

maven 依赖

<dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.15</version>
        </dependency>
        <!--  <dependency>
              <groupId>org.springframework.boot</groupId>
              <artifactId>spring-boot-starter-data-elasticsearch</artifactId>
          </dependency>-->

        <dependency>
            <groupId>org.elasticsearch.client</groupId>
            <artifactId>elasticsearch-rest-high-level-client</artifactId>
            <version>7.13.4</version>
        </dependency>
    </dependencies>

ElasticSearchConfig配置类

@Configuration
@Data
public class ElasticSearchConfig {

    @Bean
    public RestHighLevelClient restHighLevelClient() {
        return new RestHighLevelClient(
                RestClient.builder(new HttpHost("localhost", 9200, "http"))
        );
    }
}

User类

@Data
@AllArgsConstructor
@NoArgsConstructor
@ToString
public class User {
    private String name;
    private Integer age;
    private Double salary;
}

二、ES常用操作

创建es索引库(相当于创建MySQL库)

# 创建索引库
PUT userinfos
{
    "settings": {
        "number_of_shards": 3,
        "number_of_replicas": 2
      }
}

结果如下

{
  "acknowledged" : true,
  "shards_acknowledged" : true,
  "index" : "userinfos"
}

Java

# 创建索引库
PUT userinfos
{
    "settings": {
        "number_of_shards": 3,
        "number_of_replicas": 2
      }
}

	@Test
    void testCreateIndex() throws IOException {
        // 索引名 userinfos
        String index = "userinfos";
        GetIndexRequest getRequestExit = new GetIndexRequest(index);
        boolean exists = restHighLevelClient.indices().exists(getRequestExit, RequestOptions.DEFAULT);
        if (exists){
            System.out.println(index + "索引库已经存在!");
            return;
        }

        CreateIndexRequest createIndexRequest = new CreateIndexRequest(index);
        Map<String, Object> setMapping = new HashMap<>();
        // 分区数、副本数、缓存刷新时间
        setMapping.put("number_of_shards", 4);
        setMapping.put("number_of_replicas", 4);
        setMapping.put("refresh_interval", "10s");
        createIndexRequest.settings(setMapping);
        CreateIndexResponse createIndexResponse = restHighLevelClient.indices().create(createIndexRequest, RequestOptions.DEFAULT);
        boolean falg = createIndexResponse.isAcknowledged();
        if (falg) {
            System.out.println("创建索引库:" + index + "成功！");
        }
    }

删除索引

# 删除
DELETE userinfos

结果

{
  "acknowledged" : true
}

Java

    @Test
    void testDeleteIndex() throws IOException {
        DeleteIndexRequest request = new DeleteIndexRequest("userinfos");
        // 删除
        AcknowledgedResponse delete = client.indices().delete(request, RequestOptions.DEFAULT);
        System.out.println(delete.isAcknowledged());
    }

新增数据

老版本(7.0之前)

PUT /userinfos/user/1
{
  "id":"1",
  "name":"张三",
  "age":18,
  "salary":10000.00
}

结果如下

#! Deprecation: [types removal] Specifying types in document index requests is deprecated, use the typeless endpoints instead (/{index}/_doc/{id}, /{index}/_doc, or /{index}/_create/{id}).
{
  "_index" : "userinfos",
  "_type" : "user",
  "_id" : "1",
  "_version" : 1,
  "result" : "created",
  "_shards" : {
    "total" : 2,
    "successful" : 1,
    "failed" : 0
  },
  "_seq_no" : 0,
  "_primary_term" : 1
}

新版本

PUT /userinfos/_doc/1
{
  "id":"1",
  "name":"张三",
  "age":18,
  "salary":10000.00
}

结果

{
  "_index" : "userinfos",
  "_type" : "_doc",
  "_id" : "1",
  "_version" : 1,
  "result" : "created",
  "_shards" : {
    "total" : 2,
    "successful" : 1,
    "failed" : 0
  },
  "_seq_no" : 0,
  "_primary_term" : 1
}

Java

    // 测试添加文档
    @Test
    void testAddDocument() throws IOException {
        User user = new User("小兔子", 3,1000.00);
        IndexRequest request = new IndexRequest("userinfos");
        // 规则 put /userinfos/_doc/1
        request.id("3");
        request.timeout(TimeValue.timeValueSeconds(1));
        request.timeout("1s");
        request.source(JSON.toJSONString(user), XContentType.JSON);
        IndexResponse indexResponse = client.index(request, RequestOptions.DEFAULT);
        System.out.println(indexResponse.toString());
    }

Java批量新增

@Test
    void testBulkRequest() throws IOException {
        BulkRequest bulkRequest = new BulkRequest();
        bulkRequest.timeout("10s");
        ArrayList<User> userList = new ArrayList<>();
        userList.add(new User("cmolong1", 4,99.99));
        userList.add(new User("cmolong2", 5,88.88));
        userList.add(new User("cmolong3", 6,77.77));
        for (int i = 0; i < userList.size(); i++) {
            //批处理请求:批量更新和批量删除，修改对应的请求就可以了
            bulkRequest.add(new IndexRequest("userinfos").id("" + (i + 2))
                    .source(JSON.toJSONString(userList.get(i)), XContentType.JSON));
        }
        BulkResponse bulkResponse = client.bulk(bulkRequest, RequestOptions.DEFAULT);
        // bulkResponse.hasFailures() !!!!!!!  是否失败，返回 false 代表 成功 !!!!
        System.out.println(bulkResponse.hasFailures());
    }

查询数据

查询索引库 (相当于数据库的DESCRIBE user操作)

GET userinfos

结果

{
  "userinfos" : {
    "aliases" : { },
    "mappings" : {
      "properties" : {
        "age" : {
          "type" : "long"
        },
        "id" : {
          "type" : "text",
          "fields" : {
            "keyword" : {
              "type" : "keyword",
              "ignore_above" : 256
            }
          }
        },
        "name" : {
          "type" : "text",
          "fields" : {
            "keyword" : {
              "type" : "keyword",
              "ignore_above" : 256
            }
          }
        },
        "salary" : {
          "type" : "float"
        }
      }
    },
    "settings" : {
      "index" : {
        "creation_date" : "1627723442856",
        "number_of_shards" : "1",
        "number_of_replicas" : "1",
        "uuid" : "OVRZ2K4tRaS4tXGvGKi7YA",
        "version" : {
          "created" : "7060299"
        },
        "provided_name" : "userinfos"
      }
    }
  }
}

查询索引库–老版本 (相当于数据库的select * from user where id = 1操作)

GET /userinfos/user/1

结果

#! Deprecation: [types removal] Specifying types in document get requests is deprecated, use the /{index}/_doc/{id} endpoint instead.
{
  "_index" : "userinfos",
  "_type" : "user",
  "_id" : "1",
  "_version" : 1,
  "_seq_no" : 0,
  "_primary_term" : 1,
  "found" : true,
  "_source" : {
    "id" : "1",
    "name" : "张三",
    "age" : 18,
    "salary" : 10000.0
  }
}

查询索引库–新版本 (相当于数据库的select * from user where id = 1操作)

GET /userinfos/_doc/1

结果

{
  "_index" : "userinfos",
  "_type" : "_doc",
  "_id" : "1",
  "_version" : 1,
  "_seq_no" : 0,
  "_primary_term" : 1,
  "found" : true,
  "_source" : {
    "id" : "1",
    "name" : "张三",
    "age" : 18,
    "salary" : 10000.0
  }
}

Java

    // 获取文档前，优先判断是否存在 get /index/_doc/1
    @Test
    void testIsExists() throws IOException {
        GetRequest getRequest = new GetRequest("userinfos", "4");
        // 不获取返回的 _source 的上下文了
        getRequest.fetchSourceContext(new FetchSourceContext(false));
        getRequest.storedFields("_none_");
        boolean exists = client.exists(getRequest, RequestOptions.DEFAULT);
        System.out.println(exists);
    }

    // 获得文档的信息
    @Test
    void testGetDocument() throws IOException {
        GetRequest getRequest = new GetRequest("userinfos", "1");
        GetResponse getResponse = client.get(getRequest, RequestOptions.DEFAULT);
        String sourceAsString = getResponse.getSourceAsString();
        User user = JSON.parseObject(sourceAsString, User.class);
        // 转化User后的内容
        System.out.println(user);
        // 打印文档的内容
        System.out.println(sourceAsString);
        // 返回的全部内容和命令式一样的
        System.out.println(getResponse);
    }

修改数据(更新一个字段不影响其他字段)

老版本修改 (相当于数据库 update user set name = ‘张三up’ where id = 1)

POST /userinfos/user/1/_update
{
  "doc":{
   "name":"张三up"
  }
}

结果

#! Deprecation: [types removal] Specifying types in document update requests is deprecated, use the endpoint /{index}/_update/{id} instead.
{
  "_index" : "userinfos",
  "_type" : "user",
  "_id" : "1",
  "_version" : 6,
  "result" : "updated",
  "_shards" : {
    "total" : 2,
    "successful" : 1,
    "failed" : 0
  },
  "_seq_no" : 6,
  "_primary_term" : 1
}

新版本修改 (相当于数据库 update user set name = ‘张三up’ where id = 1)

POST /userinfos/_update/1
{
  "doc":{
   "name":"张三up"
  }
}

结果

{
  "_index" : "userinfos",
  "_type" : "_doc",
  "_id" : "1",
  "_version" : 2,
  "result" : "updated",
  "_shards" : {
    "total" : 2,
    "successful" : 1,
    "failed" : 0
  },
  "_seq_no" : 1,
  "_primary_term" : 1
}

Java

   // 更新文档的信息
    @Test
    void testUpdateRequest() throws IOException {
        UpdateRequest updateRequest = new UpdateRequest("userinfos", "1");
        updateRequest.timeout("1s");
        User user = new User("法外狂徒张三", 500,1000.22);
        updateRequest.doc(JSON.toJSONString(user), XContentType.JSON);
        UpdateResponse updateResponse = client.update(updateRequest, RequestOptions.DEFAULT);
        System.out.println(updateResponse.status());
    }

删除文档

DELETE userinfos/_doc/3

结果

{
  "_index" : "userinfos",
  "_type" : "_doc",
  "_id" : "3",
  "_version" : 2,
  "result" : "deleted",
  "_shards" : {
    "total" : 3,
    "successful" : 1,
    "failed" : 0
  },
  "_seq_no" : 3,
  "_primary_term" : 1
}

Java

    // 删除文档记录
    @Test
    void testDeleteRequest() throws IOException {
        DeleteRequest request = new DeleteRequest("userinfos", "3");
        request.timeout("1s");
        DeleteResponse deleteResponse = client.delete(request, RequestOptions.DEFAULT);
        System.out.println(deleteResponse.status());
    }

其他查询

{
  "_index" : "userinfos",
  "_type" : "_doc",
  "_id" : "3",
  "_version" : 2,
  "result" : "deleted",
  "_shards" : {
    "total" : 3,
    "successful" : 1,
    "failed" : 0
  },
  "_seq_no" : 3,
  "_primary_term" : 1
}

Java

    // 删除文档记录
    @Test
    void testDeleteRequest() throws IOException {
        DeleteRequest request = new DeleteRequest("userinfos", "3");
        request.timeout("1s");
        DeleteResponse deleteResponse = client.delete(request, RequestOptions.DEFAULT);
        System.out.println(deleteResponse.status());
    }