当前位置：首页 > news >正文

分布式搜索引擎Elasticsearch（一）

news 2024/11/26 12:53:07

一、初识Elasticsearch

1. 概述

elasticsearch (简称 ES) 是一款非常强大的开源的分布式的搜索引擎，可以帮助我们从海量数据中快速找到需要的内容。elasticsearch结合kibana、Logstash、Beats，也就是elastic stack（ELK），换句话说 ELK 就是以 Elasticsearch 为核心的技术栈，包括了 kibana、Logsash、Beats。ELK 被广泛应用在日志数据分析、实时监控等领域。elasticsearch是elastic stack的核心，负责存储、搜索、分析数据。

2. Elasticsearch发展

Lucene 是一个Java语言的搜索引擎类库，是Apache公司的顶级项目，由DougCutting于1999年研发。Lucene 是Apache的开源搜索引擎类库，提供了搜索引擎的核心API。

官网地址：Apache Lucene - Welcome to Apache Lucene 。

Lucene的优势：易扩展、高性能（基于倒排索引）

Lucene的缺点：只限于Java语言开发、学习曲线陡峭、不支持水平扩展

2004年Shay Banon基于Lucene开发了Compass

2010年Shay Banon 重写了Compass，取名为Elasticsearch。

官网地址: Elastic — 搜索 AI 公司 | Elastic

相比与 lucene，elasticsearch具备下列优势：

1. 支持分布式，可水平扩展。

2. 提供Restful接口，可被任何语言调用。

3. 正向索引与倒排索引

传统数据库（如MySQL）采用正向索引，即全表匹配。在局部内容检索时性能差。

elasticsearch采用倒排索引，它会重新生成一个表，里面有词条和文档id两个字段。

文档（document）：每条数据就是一个文档。
词条（term）：文档按照语义分成的词语，每个词语就是词条。

注意：如果词条已在数据表中，则不插入，确保每个词条在数据表中是唯一的。

倒排索引查询数据

如搜索”华为手机“，得到的词条就是：”华为“、”手机“，得到每个词条对应的数据 id，也就是下图的手机的文档 id 是1，2；华为的文档 id 是 2，3。接着看到 id 为 2 在两个词条中都出现了，因此优先查询 id 为 2 的数据。然后根据文档 id 查询文档，然后把数据存入结果集种。

4. ES 与 MySQL 的关系

elasticsearch是面向文档存储的，可以是数据库中的一条商品数据，一个订单信息。

文档数据会被序列化为json格式后存储在elasticsearch中。

概念对比

架构应用

Mysql：擅长事务类型操作，可以确保数据的安全和一致性。

Elasticsearch：擅长海量数据的搜索、分析、计算。

因此，ES并不是替代MySQL , 它们是互补的。在以后的开发中，看需求选择对应的架构。

5. 部署单点模式 ES 和 kibanb

部署ES:

步骤1：创建网络

因为我们还需要部署Kibana容器，因此需要让 es 和 Kibana 容器互联，这里先创建一个网络。

docker network create es-net

步骤2：下载 elasticsearch 镜像，可以到docker hub上拉取一个，由于镜像体积大，我这里直接采用本地加载的方式。

docker load -i es.tar

步骤3：运行

运行docker 命令，部署单点es:

docker run -d \--name es \-e "ES_JAVA_OPTS=-Xms512m -Xmx512m" \-e "discovery.type=single-node" \-v es-data:/usr/share/elasticsearch/data \-v es-plugins:/usr/share/elasticsearch/plugins \--privileged \--network es-net \-p 9200:9200 \-p 9300:9300 \
elasticsearch:7.12.1

命令解释：

-e "cluster.name=es-docker-cluster"`：设置集群名称
-e "http.host=0.0.0.0"`：监听的地址，可以外网访问
-e "ES_JAVA_OPTS=-Xms512m -Xmx512m"`：内存大小
-e "discovery.type=single-node"`：非集群模式
-v es-data:/usr/share/elasticsearch/data`：挂载逻辑卷，绑定es的数据目录
-v es-logs:/usr/share/elasticsearch/logs`：挂载逻辑卷，绑定es的日志目录
-v es-plugins:/usr/share/elasticsearch/plugins`：挂载逻辑卷，绑定es的插件目录
--privileged`：授予逻辑卷访问权
--network es-net` ：加入一个名为es-net的网络中
-p 9200:9200`：端口映射配置

步骤4：在浏览器中输入：http://192.168.30.130:9200 （服务器IP+ES端口）即可看到elasticsearch的响应结果：

如果容器关闭了，需要先启动容器，再启动这些服务

systemctl status docker    # 查看docker 容器状态
systemctl start docker     # 启动docker容器
docker ps    # 查看运行的服务
docker start 服务名， 如 docker start es，启动es

部署kibanb

步骤1：下载 kibanb镜像，可以到docker hub上拉取一个，由于镜像体积大，我这里直接采用本地加载的方式。

注意：kibanb版本要和ES版本一致

docker load -i kibanb.tar

步骤2：运行

运行docker 命令, 部署kibana:

docker run -d \
--name kibana \
-e ELASTICSEARCH_HOSTS=http://es:9200 \
--network=es-net \
-p 5601:5601  \
kibana:7.12.1

命令解释

--network= es-net ：加入一个名为es-net的网络中，与elasticsearch在同一个网络中
-e ELASTICSEARCH_HOSTS=http://es:9200"`：设置elasticsearch的地址，因为kibana已经与elasticsearch在一个网络，因此可以用容器名直接访问elasticsearch
-p 5601:5601`：端口映射配置

步骤3：在浏览器中输入：http://192.168.30.130:5601（服务器IP+elastic端口）即可看到管理页面：

6. 安装IK分词器

es在创建倒排索引时需要对文档分词；在搜索时，需要对用户输入内容分词。但默认的分词规则对中文处理并不友好。

处理中文分词，一般会使用IK分词器。https://github.com/medcl/elasticsearch-analysis-ik

ik分词器包含两种模式：

ik_smart：最少切分，粗粒度

ik_max_word：最细切分，细粒度

安装IK分词器：

方式一：在线安装

步骤1：进入容器内部

docker exec -it elasticsearch /bin/bash

步骤2：在线安装

./bin/elasticsearch-plugin  install https://github.com/medcl/elasticsearch-analysis-ik/releases/download/v7.12.1/elasticsearch-analysis-ik-7.12.1.zip

步骤3：退出

exit

步骤4：重启容器

docker restart elasticsearch

方式二：离线安装

步骤1：查看数据卷目录

安装插件需要知道elasticsearch的plugins目录位置，而我们用了数据卷挂载，因此需要查看elasticsearch的数据卷目录，通过下面命令查看:

docker volume inspect es-plugins

说明plugins目录被挂载到了：/var/lib/docker/volumes/es-plugins/_data 这个目录中。

步骤2：上传分词器安装包到elasticsearch的plugins目录位置，我们挂载到了/var/lib/docker/volumes/es-plugins/_data 这个目录中，解压缩分词器安装包到 ik 。

cd /var/lib/docker/volumes/es-plugins/_data    #切换到目录
mkdir ik     # 创建ik文件夹
uzip -zxvf -d /var/lib/docker/volumes/es-plugins/_data/ik elasticsearch-analysis-ik-7.12.1.zip

步骤3：重启容器

docker restart es

步骤4：查看es日志

docker logs -f es

步骤5: 测试，在浏览器中输入：http://192.168.30.130:5601（服务器IP+elastic端口）。在 Dev Tools 查看下面两种方式。

GET /_analyze
{"analyzer": "ik_smart",   #ik_smart：最少切分"text": "猫猫不是喵喵"
}
GET /_analyze
{"analyzer": "ik_max_word",   #ik_max_word：最细切分"text": "猫猫不是喵喵"
}

7. IK分词器——扩展词库

如果仔细的测试，会发现有些是词语，但分词器却划分成了单个词，有些助词，如”的，得“，无意义就不想存入占用空间。有些敏感词汇，想要屏蔽。可以使用拓展ik分词器的词库，只需要修改一个ik分词器目录中的config目录中的 IkAnalyzer.cfg.xml 文件。

config目录文件如下

IkAnalyzer.cfg.xml 文件内容如下

如果config文件目录下没有扩展字典或屏蔽词的文件名，需要自行在config目录下创建。

创建文件名为 wxt.dic 的文件，在里面编写扩展词汇

touch wxt.dic

编写扩展的词汇

重启 es 服务

docker restart es

已经看到 ”喵喵“ 被合成了一个词。

在 stopwrod.dic 的文件中编写屏蔽词汇

在 stopwrod.dic 文件中添加屏蔽词，”喵喵“。重启 es 服务

docker restart es

"喵喵" 果然被屏蔽了。

二、索引库操作

1. Mapping 映射属性

mapping是对索引库中文档的约束，常见的mapping属性包括：

1. type：字段数据类型，常见的简单类型有：

字符串：

text（可分词的文本）需要拆分、

keyword（精确值，例如：品牌、国家、ip地址）不需要拆分。

数值：long、integer、short、byte、double、float
布尔：boolean
日期：date
对象：object

2. index: 是否创建索引，默认为 true。字段需要搜索时设为true, 反之为false。

3. analyzer: 使用哪种分词器，和text结合使用。

分词器的分类：

ik_smart：最少切分，粗粒度

ik_max_word：最细切分，细粒度

4. properites：该字段的子字段

2. 索引库的CRUD

2.1 创建索引库

ES中通过Restful请求操作索引库、文档。请求内容用DSL语句来表示。创建索引库和mapping的DSL语法如下：

2.2 删除索引库

DELETE /索引库名

如 DELETE /height

2.3 修改索引库

索引库和mapping一旦创建无法修改，但是可以添加新的字段，语法如下：

PUT /索引库名/_mapping

{

"properties": {

"新字段名": {

……

}

}

}

注意：必须是全新的字段名，如果字段名存在，就会认为是修改已有的字段，会报错。

2.4 查询索引库

GET /索引库名

如 GET /height

三、文档操作

1. 新增文档

POST /索引库名/_doc/文档id

{

    "字段1": "值1",

    "字段2": "值2",

    "字段3": {

        "子属性1": "值3",

        "子属性2": "值4"

    },

    // ...

}

如

# 插入文档
POST /height/_doc/1
{
"info": "张三是法外狂徒",
"email": "zs@qq.com",
"name": {
"firstName": "三",
"lastaName": "张"
}
}

2. 删除文档

DELETE /文档名/_doc/文档id

如

# 删除文档
DELETE /height/_doc/1

3. 修改文档

方式一: 全量修改，删除旧文档，添加新文档。（全局修改）

PUT /索引库名/_doc/文档id

{

    "字段1": "值1",

    "字段2": "值2",

    // ... 略

}

如：

#修改文档（全局）
PUT /height/_doc/1
{
"info": "张三是法外狂徒666",
"email": "zs@qq.com",
"name": {
"firstName": "三",
"lastaName": "张"
}
}

方式二：增量修改，修改指定字段值。（局部修改）

POST /索引库名/_update/文档id

{

    "doc": {

         "字段名": "新的值",

    }

}

如

#修改文档（局部）
POST /height/_update/1
{
"doc": {
"email": "zs666@qq.com"
}
}

4. 查询文档

GET /文档名/_doc/文档id

如

# 查询文档
GET /height/_doc/1

5. 扩展思考

如果添加的数据没有对应设置的mapping，ES会报错吗？

答案是不会的，ES会根据我们添加的数据类型，自动为我们的字段设置mapping。如果默认的mapping设置的不符合预期，需要自行设置。

JSON类型	Elasticserch类型
字符串	日期格式字符串：mapping 设置为 date 类型普通字符串：mapping设置为text类型，并添加keyword类型子字段
布尔值	boolean
浮点数	float
整数	long
对象嵌套	object , 并添加properties
数组	由数组中第一个非空类型决定
空值	会略

mapping 设置

PUT /height 
{"mappings": {"properties": {"info": {"type": "text","analyzer": "ik_smart"},"email": {"type": "keyword","index": false},"name": {"properties": {"firstName": {"type": "keyword"},"lastName": {"type": "keyword"}}}}}
}

文档添加

#修改文档（全局）
PUT /height/_doc/1
{"info": "张三是法外狂徒666","email": "zs@qq.com","name": {"firstName": "三","lastaName": "张"},"age": 35,"score": [98.5, 99.3, 85.3],"isMarried": false,"birthday": "2000-5-6","city": "北京"
}

查询文档： GET /height/_doc/1

JSON类型

四、RestClinet

ES官方提供了各种不同语言的客户端，用来操作ES。这些客户端的本质就是组装DSL语句，通过http请求发送给ES。官方文档地址：https://www.elastic.co/guide/en/elasticsearch/client/index.html

对于地理坐标这种特殊的值，ES给出了2种坐标数据类型：

geo_point：由纬度和经度确定一个点。如：”32.8752345, 120.2981576"

geo_shape：由多个geo_point 组成的复杂集合图。如一条直线。

字段拷贝可以使用copy_to属性将当前字段拷贝到指定字段

例如：把brand字段拷贝到 all字段中。

"all": {"type": "text","analyzer": "ik_max_word"
},
"brand": {"type": "keyword","copy_to": "all"
}

一）RestClient 操作索引库

1. 初始化JavaRestClient

1.1. 引入es的RestHighLevelClient依赖

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

1.2. SpringBoot默认的ES版本是7.6.2，所以我们需要覆盖默认的ES版本：

<properties>   <java.version>1.8</java.version>    <elasticsearch.version>7.12.1</elasticsearch.version> 
</properties>

覆盖后的效果

1.3. 初始化RestHighLevelClient：

public class HotelIndexTest {private RestHighLevelClient client;// 客户端初始化@BeforeEachvoid setUp() {this.client = new RestHighLevelClient(RestClient.builder(HttpHost.create("https://192.168.30.130:9200")  // 服务器IP + ES 端口));}@Testvoid testInit() {System.out.println(client);}@AfterEachvoid tearDown() throws IOException {this.client.close();}
}

2. 创建索引库

定义一个类存放常量 DSL 语句

package cn.itcast.hotel.constants;
/**
*  定义常量存放 DSL
* */
public class HotelConstants {public static final String MAPPING_TEMPLATE = "{\n" +"  \"mappings\": {\n" +"    \"properties\": {\n" +"      \"id\": {\n" +"        \"type\": \"keyword\"\n" +"      },\n" +"      \"name\": {\n" +"        \"type\": \"text\",\n" +"        \"analyzer\": \"ik_max_word\",\n" +"        \"copy_to\": \"all\"\n" +"      },\n" +"      \"address\": {\n" +"        \"type\": \"keyword\",\n" +"        \"index\": false\n" +"      },\n" +"      \"price\": {\n" +"        \"type\": \"integer\"\n" +"      },\n" +"      \"score\": {\n" +"        \"type\": \"integer\"\n" +"      },\n" +"      \"brand\": {\n" +"        \"type\": \"keyword\",\n" +"        \"copy_to\": \"all\"\n" +"      },\n" +"      \"city\": {\n" +"        \"type\": \"keyword\"\n" +"      },\n" +"      \"startName\": {\n" +"        \"type\": \"keyword\"\n" +"      },\n" +"      \"business\": {\n" +"        \"type\": \"keyword\",\n" +"        \n" +"        \"copy_to\": \"all\"\n" +"      },\n" +"      \"location\": {\n" +"        \"type\": \"geo_point\"\n" +"      },\n" +"      \"pic\": {\n" +"        \"type\": \"keyword\",\n" +"        \"index\": false\n" +"      },\n" +"      \"all\": {\n" +"        \"type\": \"text\",\n" +"        \"analyzer\": \"ik_max_word\"\n" +"      }\n" +"\n" +"    }\n" +"  }\n" +"}";
}

添加如下代码块，Java 与 DSL 建立连接

    @Testvoid testCreateHotelIndex() throws IOException {// 1.创建Request对象CreateIndexRequest request = new CreateIndexRequest("hotel");// 2.请求参数，MAPPING_TEMPLATE是静态常量字符串， 内容是创建索引库的DSL语句request.source(MAPPING_TEMPLATE, XContentType.JSON);// 3.发起请求client.indices().create(request, RequestOptions.DEFAULT);}

整体代码

import org.apache.http.HttpHost;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.common.xcontent.XContentType;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;import java.io.IOException;import static cn.itcast.hotel.constants.HotelConstants.MAPPING_TEMPLATE;public class HotelIndexTest {private RestHighLevelClient client;// 客户端初始化@BeforeEachvoid setUp() {this.client = new RestHighLevelClient(RestClient.builder(HttpHost.create("http://192.168.30.130:9200")  // 服务器IP + ES 端口));}@Testvoid testInit() {System.out.println(client);}@Testvoid testCreateHotelIndex() throws IOException {// 1.创建Request对象CreateIndexRequest request = new CreateIndexRequest("hotel");// 2.请求参数，MAPPING_TEMPLATE是静态常量字符串， 内容是创建索引库的DSL语句request.source(MAPPING_TEMPLATE, XContentType.JSON);// 3.发起请求client.indices().create(request, RequestOptions.DEFAULT);}@AfterEachvoid tearDown() throws IOException {this.client.close();}}

3. 删除索引库

    @Testvoid testDeleteHotelIndex() throws IOException {// 1.创建Request对象DeleteIndexRequest request = new DeleteIndexRequest("hotel");// 2.发起请求client.indices().delete(request, RequestOptions.DEFAULT);}

4. 判断索引库是否存在

  @Testvoid testExistsHotelIndex() throws IOException {// 1.创建Request对象GetIndexRequest request = new GetIndexRequest("hotel");// 2.发起请求boolean exists = client.indices().exists(request, RequestOptions.DEFAULT);// 3.输出System.err.println(exists ? "索引库已存在" : "索引库不存在");}

二）RestClient操作文档

1. 初始化JavaRestClient

对文档增删查改时，需要先初始化JavaRestClient,操作步骤和上面的操作索引库一致。这样不在重复描述。

2. 新增操作文档

@SpringBootTest
public class HotelDocumentTest {private RestHighLevelClient client;@Autowiredprivate IHotelService hotelService;// 客户端初始化@BeforeEachvoid setUp() {this.client = new RestHighLevelClient(RestClient.builder(HttpHost.create("http://192.168.30.130:9200")  // 服务器IP + ES 端口));}@Testvoid testAddDocument() throws IOException {// 根据id查询数据Hotel hotel = hotelService.getById(696948L);// 转换文档类型HotelDoc hotelDoc = new HotelDoc(hotel);// 1.创建Request对象IndexRequest request = new IndexRequest("hotel").id(hotel.getId().toString());// 2.请求JSON文档request.source(JSON.toJSONString(hotelDoc), XContentType.JSON);// 3.发起请求client.index(request, RequestOptions.DEFAULT);}@AfterEachvoid tearDown() throws IOException {this.client.close();}}

3. 根据id查询文档

根据id查询到的文档数据是json，需要反序列化为java对象：

@SpringBootTest
public class HotelDocumentTest {private RestHighLevelClient client;@Autowiredprivate IHotelService hotelService;// 客户端初始化@BeforeEachvoid setUp() {this.client = new RestHighLevelClient(RestClient.builder(HttpHost.create("http://192.168.30.130:9200")  // 服务器IP + ES 端口));}@Testvoid testGetDocument() throws IOException {// 1. 准备requestGetRequest request = new GetRequest("hotel", "696948");// 2. 发送请求，得到相应GetResponse response = client.get(request, RequestOptions.DEFAULT);//3. 解析相应结果String json = response.getSourceAsString();// 反序列化HotelDoc hotelDoc = JSON.parseObject(json, HotelDoc.class);System.out.println(hotelDoc);}@AfterEachvoid tearDown() throws IOException {this.client.close();}}

执行结果

4. 删除操作文档

@SpringBootTest
public class HotelDocumentTest {private RestHighLevelClient client;@Autowiredprivate IHotelService hotelService;// 客户端初始化@BeforeEachvoid setUp() {this.client = new RestHighLevelClient(RestClient.builder(HttpHost.create("http://192.168.30.130:9200")  // 服务器IP + ES 端口));}@Testvoid testDeleteDocument() throws IOException {// 1. 准备requestDeleteRequest request = new DeleteRequest("hotel", "696948");// 2. 发送请求client.delete(request, RequestOptions.DEFAULT);}@AfterEachvoid tearDown() throws IOException {this.client.close();}}

5. 修改操作文档

修改文档数据有两种方式：

方式一：全量更新。再次写入id一样的文档，就会删除旧文档，添加新文档

方式二：局部更新。只更新部分字段，我们演示方式二

@SpringBootTest
public class HotelDocumentTest {private RestHighLevelClient client;@Autowiredprivate IHotelService hotelService;// 客户端初始化@BeforeEachvoid setUp() {this.client = new RestHighLevelClient(RestClient.builder(HttpHost.create("http://192.168.30.130:9200")  // 服务器IP + ES 端口));}@Testvoid testUpdateDocument() throws IOException {// 1. 准备requestUpdateRequest request = new UpdateRequest("hotel", "696948");// 2. 准备请求参数request.doc(     // 注意：键值对之间使用逗号隔开的"price", "952","starName", "四钻");// 3. 发送请求client.update(request, RequestOptions.DEFAULT);}@AfterEachvoid tearDown() throws IOException {this.client.close();}}

运行结果

6. 批量导入数据到ES

批量导入数据到ES的步骤：

利用mybatis-plus查询酒店数据
将查询到的酒店数据（Hotel）转换为文档类型数据（HotelDoc）
利用JavaRestClient中的Bulk批处理，实现批量新增文档，示例代码如下

package cn.itcast.hotel;import cn.itcast.hotel.pojo.Hotel;
import cn.itcast.hotel.pojo.HotelDoc;
import cn.itcast.hotel.service.IHotelService;
import cn.itcast.hotel.service.impl.HotelService;
import com.alibaba.fastjson.JSON;
import org.apache.http.HttpHost;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.common.xcontent.XContentType;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;import java.io.IOException;
import java.util.List;import static cn.itcast.hotel.constants.HotelConstants.MAPPING_TEMPLATE;@SpringBootTest
public class HotelDocumentTest {private RestHighLevelClient client;@Autowiredprivate IHotelService hotelService;// 客户端初始化@BeforeEachvoid setUp() {this.client = new RestHighLevelClient(RestClient.builder(HttpHost.create("http://192.168.30.130:9200")  // 服务器IP + ES 端口));}@Testvoid testBulk() throws IOException {// 查询所有数据List<Hotel> hotels = hotelService.list();// 1.创建Bulk请求BulkRequest request = new BulkRequest();// 2.添加要批量提交的请求for (Hotel hotel : hotels) {// 转换为文档类型 HotelDocHotelDoc hotelDoc = new HotelDoc(hotel);request.add(new IndexRequest("hotel").id(hotelDoc.getId().toString()).source(JSON.toJSONString(hotelDoc), XContentType.JSON));}// 3.发起bulk请求client.bulk(request, RequestOptions.DEFAULT);}@AfterEachvoid tearDown() throws IOException {this.client.close();}}

批量查询