ES官方提供了各种不同语言的客户端,用来操作ES。这些客户端的本质就是组装DSL语句,通过http请求发送给ES。官方文档地址: Elasticsearch Clients | Elastic。由于ES目前最新版本是8.16,提供了全新版本的客户端,老版本的客户端已经被标记为过时。而我们采用的是7.12版本,因此只能使用老版本客户端:
1.RestAPI
1.1.初始化RestClient
在elasticsearch提供的API中,与elasticsearch一切交互都封装在一个名为RestHighLevelClient
的类中,必须先完成这个对象的初始化,建立与elasticsearch的连接。
分为两步:
(1) 导入依赖
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-elasticsearch</artifactId><version>7.12.1</version>
</dependency>
<dependency><groupId>org.elasticsearch.client</groupId><artifactId>elasticsearch-rest-high-level-client</artifactId><version>7.12.1</version>
</dependency>
(2) 初始化RestHighLevelClient
初始化的代码如下:
java">RestHighLevelClient client = new RestHighLevelClient(RestClient.builder(HttpHost.create("http://192.168.150.101:9200")
));
这里为了单元测试方便,我们创建一个测试类IndexTest
,然后将初始化的代码编写在@BeforeEach
方法中:
java">public class IndexTest {private RestHighLevelClient client;@BeforeEachvoid setUp() {this.client = new RestHighLevelClient(RestClient.builder(HttpHost.create("http://192.168.150.101:9200")));}@Testvoid testConnect() {System.out.println(client);}@AfterEachvoid tearDown() throws IOException {this.client.close();}
}
1.2.创建索引库
1.2.1.Mapping映射
Mapping是对索引库中文档的约束,常见的Mapping属性包括:
-
type
:字段数据类型,常见的简单类型有:-
字符串:
text
(可分词的文本)、keyword
(精确值,例如:品牌、国家、ip地址) -
数值:
long
、integer
、short
、byte
、double
、float
、 -
布尔:
boolean
-
日期:
date
-
对象:
object
-
-
index
:是否创建索引,默认为true
-
analyzer
:使用哪种分词器 -
properties
:该字段的子字段
下面是一个索引库案例:
java">PUT /items
{"mappings": {"properties": {"id": {"type": "keyword"},"name":{"type": "text","analyzer": "ik_max_word"},"price":{"type": "integer"},"stock":{"type": "integer"},"image":{"type": "keyword","index": false},"category":{"type": "keyword"},"brand":{"type": "keyword"},"sold":{"type": "integer"},"commentCount":{"type": "integer","index": false},"isAD":{"type": "boolean"},"updateTime":{"type": "date"}}}
}
1.2.2. 创建索引
创建索引库的API如下:
代码分为三步:
-
1)创建Request对象。
-
因为是创建索引库的操作,因此Request是
CreateIndexRequest
。
-
-
2)添加请求参数
-
其实就是Json格式的Mapping映射参数。因为json字符串很长,这里是定义了静态字符串常量
MAPPING_TEMPLATE
,让代码看起来更加优雅。
-
-
3)发送请求
-
client.
indices
()
方法的返回值是IndicesClient
类型,封装了所有与索引库操作有关的方法。例如创建索引、删除索引、判断索引是否存在等
-
在IndexTest
测试类中,具体代码如下:
java">@Test
void testCreateIndex() throws IOException {// 1.创建Request对象CreateIndexRequest request = new CreateIndexRequest("items");// 2.准备请求参数request.source(MAPPING_TEMPLATE, XContentType.JSON);// 3.发送请求client.indices().create(request, RequestOptions.DEFAULT);
}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" +" },\n" +" \"price\":{\n" +" \"type\": \"integer\"\n" +" },\n" +" \"stock\":{\n" +" \"type\": \"integer\"\n" +" },\n" +" \"image\":{\n" +" \"type\": \"keyword\",\n" +" \"index\": false\n" +" },\n" +" \"category\":{\n" +" \"type\": \"keyword\"\n" +" },\n" +" \"brand\":{\n" +" \"type\": \"keyword\"\n" +" },\n" +" \"sold\":{\n" +" \"type\": \"integer\"\n" +" },\n" +" \"commentCount\":{\n" +" \"type\": \"integer\"\n" +" },\n" +" \"isAD\":{\n" +" \"type\": \"boolean\"\n" +" },\n" +" \"updateTime\":{\n" +" \"type\": \"date\"\n" +" }\n" +" }\n" +" }\n" +"}";
1.3.删除索引库
删除索引库的请求非常简单:
java">DELETE /hotel
与创建索引库相比:
-
请求方式从PUT变为DELTE
-
请求路径不变
-
无请求参数
所以代码的差异,注意体现在Request对象上。流程如下:
-
1)创建Request对象。这次是DeleteIndexRequest对象
-
2)准备参数。这里是无参,因此省略
-
3)发送请求。改用delete方法
-
在
IndexTest
测试类中,编写单元测试,实现删除索引:
java">@Test
void testDeleteIndex() throws IOException {// 1.创建Request对象DeleteIndexRequest request = new DeleteIndexRequest("items");// 2.发送请求client.indices().delete(request, RequestOptions.DEFAULT);
}
1.4.判断索引库是否存在
判断索引库是否存在,本质就是查询,对应的请求语句是:
java">GET /hotel
因此与删除的Java代码流程是类似的,流程如下:
-
1)创建Request对象。这次是GetIndexRequest对象
-
2)准备参数。这里是无参,直接省略
-
3)发送请求。改用exists方法
java">@Test
void testExistsIndex() throws IOException {// 1.创建Request对象GetIndexRequest request = new GetIndexRequest("items");// 2.发送请求boolean exists = client.indices().exists(request, RequestOptions.DEFAULT);// 3.输出System.err.println(exists ? "索引库已经存在!" : "索引库不存在!");
}
总结
JavaRestClient操作elasticsearch的流程基本类似。核心是client.indices()
方法来获取索引库的操作对象。
索引库操作的基本步骤:
-
初始化
RestHighLevelClient
-
创建XxxIndexRequest。XXX是
Create
、Get
、Delete
-
准备请求参数(
Create
时需要,其它是无参,可以省略) -
发送请求。调用
RestHighLevelClient#indices().xxx()
方法,xxx是create
、exists
、delete
2. RestClient操作文档
索引库准备好以后,就可以操作文档了。为了与索引库操作分离,我们再次创建一个测试类,做两件事情:
-
初始化RestHighLevelClient
-
我们的商品数据在数据库,需要利用IHotelService去查询,所以注入这个接口
java">@SpringBootTest(properties = "spring.profiles.active=local")
public class DocumentTest {private RestHighLevelClient client;@Autowiredprivate IItemService itemService;@BeforeEachvoid setUp() {this.client = new RestHighLevelClient(RestClient.builder(HttpHost.create("http://192.168.150.101:9200")));}@AfterEachvoid tearDown() throws IOException {this.client.close();}
}
2.1.新增文档
我们需要将数据库中的信息导入elasticsearch中
2.1.1.实体类
索引库结构与数据库结构还存在一些差异,因此我们要定义一个索引库结构对应的实体。
java">@Data
@ApiModel(description = "索引库实体")
public class ItemDoc{@ApiModelProperty("商品id")private String id;@ApiModelProperty("商品名称")private String name;@ApiModelProperty("价格(分)")private Integer price;@ApiModelProperty("商品图片")private String image;@ApiModelProperty("类目名称")private String category;@ApiModelProperty("品牌名称")private String brand;@ApiModelProperty("销量")private Integer sold;@ApiModelProperty("评论数")private Integer commentCount;@ApiModelProperty("是否是推广广告,true/false")private Boolean isAD;@ApiModelProperty("更新时间")private LocalDateTime updateTime;
}
2.1.2.API语法
新增文档的请求语法如下:
java">POST /{索引库名}/_doc/1
{"name": "Jack","age": 21
}
对应的JavaAPI如下:
可以看到与索引库操作的API非常类似,同样是三步走:
-
1)创建Request对象,这里是
IndexRequest
,因为添加文档就是创建倒排索引的过程 -
2)准备请求参数,本例中就是Json文档
-
3)发送请求
变化的地方在于,这里直接使用client.xxx()
的API,不再需要client.indices()
了。
2.1.3.完整代码
我们导入商品数据,除了参考API模板“三步走”以外,还需要做几点准备工作:
-
商品数据来自于数据库,我们需要先查询出来,得到
Item
对象 -
Item
对象需要转为ItemDoc
对象 -
ItemDTO
需要序列化为json
格式
因此,代码整体步骤如下:
-
1)根据id查询商品数据
Item
-
2)将
Item
封装为ItemDoc
-
3)将
ItemDoc
序列化为JSON -
4)创建IndexRequest,指定索引库名和id
-
5)准备请求参数,也就是JSON文档
-
6)发送请求
在测试类中,编写单元测试:
java">@Test
void testAddDocument() throws IOException {// 1.根据id查询商品数据Item item = itemService.getById(100002644680L);// 2.转换为文档类型ItemDoc itemDoc = BeanUtil.copyProperties(item, ItemDoc.class);// 3.将ItemDTO转jsonString doc = JSONUtil.toJsonStr(itemDoc);// 1.准备Request对象IndexRequest request = new IndexRequest("items").id(itemDoc.getId());// 2.准备Json文档request.source(doc, XContentType.JSON);// 3.发送请求client.index(request, RequestOptions.DEFAULT);
}
2.2.查询文档
我们以根据id查询文档为例,查询的请求语句如下:
java">GET /{索引库名}/_doc/{id}
与之前的流程类似,代码大概分2步:
-
创建Request对象
-
发送请求
不过查询的目的是得到结果,解析为ItemDTO,还要再加一步对结果的解析。示例代码如下:
可以看到,响应结果是一个JSON,其中文档放在一个_source
属性中,因此解析就是拿到_source
,反序列化为Java对象即可。
其它代码与之前类似,流程如下:
-
1)准备Request对象。这次是查询,所以是
GetRequest
-
2)发送请求,得到结果。因为是查询,这里调用
client.get()
方法 -
3)解析结果,就是对JSON做反序列化
完整代码
java">@Test
void testGetDocumentById() throws IOException {// 1.准备Request对象GetRequest request = new GetRequest("items").id("100002644680");// 2.发送请求GetResponse response = client.get(request, RequestOptions.DEFAULT);// 3.获取响应结果中的sourceString json = response.getSourceAsString();ItemDoc itemDoc = JSONUtil.toBean(json, ItemDoc.class);System.out.println("itemDoc= " + ItemDoc);
}
2.3.删除文档
删除的请求语句如下:
java">DELETE /hotel/_doc/{id}
与查询相比,仅仅是请求方式从DELETE
变成GET
,可以想象Java代码应该依然是2步走:
-
1)准备Request对象,因为是删除,这次是
DeleteRequest
对象。要指定索引库名和id -
2)发送请求。因为是删除,所以是
client.delete()
方法
在DocumentTest
测试类中,编写单元测试:
java">@Test
void testDeleteDocument() throws IOException {// 1.准备Request,两个参数,第一个是索引库名,第二个是文档idDeleteRequest request = new DeleteRequest("item", "100002644680");// 2.发送请求client.delete(request, RequestOptions.DEFAULT);
}
2.4.修改文档
修改有两种方式:
-
全量修改:本质是先根据id删除,再新增
-
局部修改:修改文档中的指定字段值
在RestClient的API中,全量修改与新增的API完全一致,判断依据是ID:
-
如果新增时,ID已经存在,则修改
-
如果新增时,ID不存在,则新增
局部修改的请求语法如下:
java">POST /{索引库名}/_update/{id}
{"doc": {"字段名": "字段值","字段名": "字段值"}
}
代码示例如图:
与之前类似,也是三步走:
-
1)准备
Request
对象。这次是修改,所以是UpdateRequest
-
2)准备参数。也就是JSON文档,里面包含要修改的字段
-
3)更新文档。这里调用
client.update()
方法
java">@Test
void testUpdateDocument() throws IOException {// 1.准备RequestUpdateRequest request = new UpdateRequest("items", "100002644680");// 2.准备请求参数request.doc("price", 58800,"commentCount", 1);// 3.发送请求client.update(request, RequestOptions.DEFAULT);
}
2.5. 批量导入文档
在之前的案例中,我们都是操作单个文档。而数据库中的商品数据实际会达到数十万条,某些项目中可能达到数百万条。
我们如果要将这些数据导入索引库,肯定不能逐条导入,而是采用批处理方案。常见的方案有:
-
利用Logstash批量导入
-
需要安装Logstash
-
对数据的再加工能力较弱
-
无需编码,但要学习编写Logstash导入配置
-
-
利用JavaAPI批量导入
-
需要编码,但基于JavaAPI,学习成本低
-
更加灵活,可以任意对数据做再加工处理后写入索引库
-
接下来,我们就学习下如何利用JavaAPI实现批量文档导入。
2.5.1.语法说明
批处理与前面讲的文档的CRUD步骤基本一致:
-
创建Request,但这次用的是
BulkRequest
-
准备请求参数
-
发送请求,这次要用到
client.bulk()
方法
BulkRequest
本身其实并没有请求参数,其本质就是将多个普通的CRUD请求组合在一起发送。例如:
-
批量新增文档,就是给每个文档创建一个
IndexRequest
请求,然后封装到BulkRequest
中,一起发出。 -
批量删除,就是创建N个
DeleteRequest
请求,然后封装到BulkRequest
,一起发出
因此BulkRequest
中提供了add
方法,用以添加其它CRUD的请求:
可以看到,能添加的请求有:
-
IndexRequest
,也就是新增 -
UpdateRequest
,也就是修改 -
DeleteRequest
,也就是删除
因此Bulk中添加了多个IndexRequest
,就是批量新增功能了。示例:
java">@Test
void testBulk() throws IOException {// 1.创建RequestBulkRequest request = new BulkRequest();// 2.准备请求参数request.add(new IndexRequest("items").id("1").source("json doc1", XContentType.JSON));request.add(new IndexRequest("items").id("2").source("json doc2", XContentType.JSON));// 3.发送请求client.bulk(request, RequestOptions.DEFAULT);
}
当我们要导入商品数据时,由于商品数量达到数十万,因此不可能一次性全部导入。建议采用循环遍历方式,每次导入1000条左右的数据。
测试类中,编写单元测试:
java">@Test
void testLoadItemDocs() throws IOException {// 分页查询商品数据int pageNo = 1;int size = 1000;while (true) {Page<Item> page = itemService.lambdaQuery().eq(Item::getStatus, 1).page(new Page<Item>(pageNo, size));// 非空校验List<Item> items = page.getRecords();if (CollUtils.isEmpty(items)) {return;}log.info("加载第{}页数据,共{}条", pageNo, items.size());// 1.创建RequestBulkRequest request = new BulkRequest("items");// 2.准备参数,添加多个新增的Requestfor (Item item : items) {// 2.1.转换为文档类型ItemDTOItemDoc itemDoc = BeanUtil.copyProperties(item, ItemDoc.class);// 2.2.创建新增文档的Request对象request.add(new IndexRequest().id(itemDoc.getId()).source(JSONUtil.toJsonStr(itemDoc), XContentType.JSON));}// 3.发送请求client.bulk(request, RequestOptions.DEFAULT);// 翻页pageNo++;}
}
小结
文档操作的基本步骤:
-
初始化
RestHighLevelClient
-
创建XxxRequest。
-
XXX是
Index
、Get
、Update
、Delete
、Bulk
-
-
准备参数(
Index
、Update
、Bulk
时需要) -
发送请求。
-
调用
RestHighLevelClient#.xxx()
方法,xxx是index
、get
、update
、delete
、bulk
-
-
解析结果(
Get
时需要)