Elasticsearch 介绍及java集成
一、Elasticsearch 基础介绍
ElasticSearch 是分布式实时搜索、实时分析、实时存储引擎,简称(ES), 成立于2012年,是一家来自荷兰的、开源的大数据搜索、分析服务提供商,为企业提供实时搜索、数据分析服务,支持PB级的大数据。
基于Apache Lucene 开源搜索引擎,Lucene是目前公认的性能最好,最先进的,功能最全的搜索引擎。
Elasticsearch使用Java开发并使用Lucene作为其核心来实现所有索引和搜索的功能,通过简单RESTfulAPI来隐藏Lucene的复杂性,从而让全文搜索变得简单。 速度超出你的想像,从10亿的数据中查出一条只需要1-2秒
除了Lucene 和全文搜索,还有以下功能
分布式的实时文件存储,每个字段都被索引并可被搜索
分布式的实时分析搜索引擎
可以扩展到上百台服务器,处理PB级结构化或非结构化数据
而且,所有的这些功能被集成到一个服务里面,你的应用可以通过简单的RESTful API、各种语言的客户端甚至命令行与之交互。
为什么要用ElasticSearch?
全文检索开始使用SQL来写,使用like进行模糊查询。如果数据量比较大,用这种方法就会特别慢,可以使用索引使得速度相对提高,但还是达不到对大数据搜索的要求,所以要使用分布式的全文搜索引擎ElasticSearch。
1)、ES原理剖析
索引和搜索流程图
绿色代表索引过程,对要检索的内容进行索引构建一个索引库,
索引过程包括:确定的原始内容即要搜索的内容——>采集文档——>创建文档——>分析文档——>索引文档
红色代表搜索过程:从索引库中搜索内容,
搜索过程:用户通过搜索界面——>创建查询——>执行搜索,从索引库搜索——>渲染搜索结果
二、Elasticsearch基本概念:
索引(Index)
一个索引就是一个拥有几分相似特征的文档的集合。比如说,你可以有一个客户数据的索引,另一个产品目录的索引,还有一个订单数据的索引。一个索引由一个名字来标识(必须全部是小写字母的),并且当我们要对对应于这个索引中的文档进行索引、搜索、更新和删除的时候,都要使用到这个名字。
类型(Type)6.0.0版本中弃用
类型,曾经是索引的逻辑类别/分区,允许您在同一索引中存储不同类型的文档,例如,一种类型用于用户,另一种类型用于博客帖子。
在一个索引中,你可以定义一种或多种类型。一个类型是你的索引的一个逻辑上的分类/分区,其语义完全由你来定。
文档(Document)
一个文档是一个可被索引的基础信息单元。比如,你可以拥有某一个客户的文档,某一个产品的一个文档,当然,也可以拥有某个订单的一个文档。文档以JSON(Javascript Object Notation)格式来表示。文档必须被索引/赋予一个索引的type。
分片(Shards)
索引可能存储大量可能超过单个节点的硬件限制的数据。
如果我们的索引数据量很大,超过硬件存放单个文件的限制,就会影响查询请求的速度,Es引入了分片技术。一个分片本身就是一个完成的搜索引擎,文档存储在分片中,而分片会被分配到集群中的各个节点中,随着集群的扩大和缩小,ES会自动的将分片在节点之间进行迁移,以保证集群能保持一种平衡。分片有以下特点:
- ES的一个索引可以包含多个分片(shard);
- 每一个分片(shard)都是一个最小的工作单元,承载部分数据;
- 每个shard都是一个lucene实例,有完整的简历索引和处理请求的能力;
- 增减节点时,shard会自动在nodes中负载均衡;
- 一个文档只能完整的存放在一个shard上
- 一个索引中含有shard的数量,默认值为5,在索引创建后这个值是不能被更改的。
- 优点:水平分割和扩展我们存放的内容索引;分发和并行跨碎片操作提高性能/吞吐量;
- 每一个shard关联的副本分片(replica shard)的数量,默认值为1,这个设置在任何时候都可以修改。
副本(Replicasedit)
副本,是对分片的复制。目的是为了当分片/节点发生故障时提供高可用性,它允许您扩展搜索量/吞吐量,因为可以在所有副本上并行执行搜索。
一个分片可以有多个复制分片,也可以无复制分片。它的作用主要是防止分片故障,加速查询索引等功能,提供了高可用性。另外,复制分片是不和主分片在一起的,一个主分片在一台机器上,它的复制分片可能分布在其它N台机器上。在这里,我们可以把它理解为,一个分片的复制,就叫复制分片。每个分片会包含部分索引文件。文件由sgment组成 。
副本(replica shard)就是shard的冗余备份,它的主要作用:
1)、冗余备份,防止数据丢失;
2)、shard异常时负责容错和负载均衡;
注意:副本是乘法,越多越浪费,但也越保险。分片是除法,分片越多,单分片数据就越少也越分散。
集群
多台ES服务器的结合的统称叫ES集群,一个集群包含多台服务器,多个节点。
节点
一个节点是你集群中的一个服务器,作为集群的一部分,它存储你的数据,参与集群的索引和搜索功能。
节点种类
主节点:负责集群范围内轻量级的操作,例如创建或删除索引。跟踪那些节点是集群的一部分以及确定将哪些碎片分配给哪些节点
数据节点:包含已创建的索引文档的分片。数据节点处理及数据相关的操作。例如CRUD,搜索和聚合
调节节点:仅可路由请求,处理搜索缩减阶段并分配批量索引。本质上,仅协调节点充当智能负载平衡器
节点和分片如何工作?
一个集群至少有一个节点,而一个节点就是一个ElasticSearch进程,节点可以有多个默认索引,如果创建索引,索引将会由5个分片(primary shard,又称主分片)构成,每一个分片会有一个复制分片
三、与传统的关系型数据库中的库、表、行、列等概念进行对比
关系型数据库 -> Databases(库) -> Tables(表) -> Rows(行) -> Columns(列)。
Elasticsearch -> Indeces(索引) -> Types(类型) -> Documents(文档) -> Fields(属性)。
RDBS |
ES |
数据库(database) |
索引(index) |
表(table) |
类型(type)(ES6.0之后被废弃,es7中完全删除) |
表结构(schema) |
映射(mapping) |
行(row) |
文档(document) |
列(column) |
字段(field) |
索引(Schema) |
反向索引(Mapping) |
SQL |
查询DSL |
SELECT * FROM table |
GET http://..... |
UPDATE table SET |
PUT http://...... |
DELETE |
DELETE http://...... |
(1)、关系型数据库中的数据库(database),等价与ES索引(index)
(2)、一个数据库下面有N张表(table),等价与1个索引Index下面有N多类型(Type)
备注:(ES6.0之后被废弃,es7中完全删除)
(3)、一个数据库表(table)下的数据有多行(row)多列(colum)组成,等价与一个Type由多文档(document)多字段(field)组成
(4)、在一个关系型数据库中,索引(Schema)定义了表,每个表的字段,还有表和字段的之间关系,与之对应,在ES中:反向索引(Mapping)定义索引下的Ttype的字段的处理规则,即如何建立、索引类型、是否保存原始索引JSON文档、是否压缩原始JSON文档、是否需要分词处理、如何进行分词处理等
(5)、在数据库中新增 INSERT、删除 DELTE、修改 UPDATE、查询 SEARCH操作等价于ES中的新增PUT/POST、删除DELETE、修改_update、查询GET
ES内置的RREST接口
搜索原理
(1)、客户端给DODE1发送请求,查询名字叫张三的数据
(2)、P1节点接收到请求,判断出当前数据的_ID对应的分片0,且分片P1中的数据对应复制分片R0,R1都有,就会将请求转发到R0进行处理
(3)、取出文档数据返回给P1,最终返回给前端
更新原理
(1)、客户端给NODE1发送更新请求
(2)、它转发请求到主分片所在的节点NODE3
(3)、NODE3从主分片检索出文档,修改_soure字段的JSON,然后在主分片上重建索引,如果有其他进程修改了文档,它以retry_on_conflict设置的次数重复步骤3,都未成功则放弃
(4)、如何NODE3更新文档成功,它同时转发文档的新版本到NODE1和NODE2上的复制节点以重建索引。当所有复制节点更新成功,NODE3返回成功给请求节点,然后返回用户端
创建/删除原理
(1)、客户端发送请求创建、删除请求
(2)、根据文档ID,它将转发请求到主分片所在节点NODE3
(3)、NODE3在主分片上执行请求,如果成功将转发请求到NODE1和NODE2的复制分片上,当所有复制分片成功,则NODE3返回成功信息给请求节点。在将信息返回给客户端
字段数据类型:
字符型:text(分词,不能用于排序、过滤查询、聚合查询)、keyWord
数字型:byte、short、integer、float、double
布尔型:boolean
日期型:date
二进制型:binary
对象类型:object
字段属性:
store:是否储存字段原始值(独立于_source字段)
index:是否参与索引
analyzer:指定分词器
boost:字段级别的分数加权
doc_values:是否对不分词建立正排序索引
fleld_data:是否对分词器建立正排序索引
properties:类型映射
ignore_above:超过指定字符的文本将忽略不被索引
include_in_all:是否包含该字段到_all字段中
index_optionss:倒排序索引的可选参数
norms:是否储存长度因子和分数加权(boost)
null_value:初始值
position_increment_gap:指定多字段的多个值之间的位置间隔
search_analyzer:指定搜索时分词器
similarity:指定评分策略
term_vector:指定返回哪些关于词条的统计信息
normalizer:标注化处理器
coerce:强制类型转换器
copy_to:创建自定义的_all属性
dynamic:动态映射策略
enabled:是否处理字段(正排序索引和倒排序索引)
eager_global_ordinals:是否立即加载全局序号
format:指定日期格式
ignore_malformed:忽略格式错误的字段
四、ES的特性:
速度快、易扩展、弹性、灵活、操作简单、多语言客户端、X-Pack、hadoop/spark强强联手、开箱即用。
- 分布式:横向扩展非常灵活
- 全文检索:基于lucene的强大的全文检索能力;
- 近实时搜索和分析:数据进入ES,可达到近实时搜索,还可进行聚合分析
- 高可用:容错机制,自动发现新的或失败的节点,重组和重新平衡数据
- 模式自由:ES的动态mapping机制可以自动检测数据的结构和类型,创建索引并使数据可搜索。
- RESTful API:JSON + HTTP
五、索引的应用
创建索引
PUT project_v1
{
"settings": {
"number_of_shards": 1,
"number_of_replicas": 1
},
"mappings": {
"properties": {
"name_cn": {
"type": "text"
},
"name_en": {
"type": "keyword"
},
"project_type": {
"type": "keyword"
},
"people_count": {
"type": "integer"
},
"order_count": {
"type": "long"
},
"date": {
"type": "date",
"format": "yyyy-MM-dd HH:mm:ss||yyyy-MM-dd||yyyy-MM||epoch_millis"
}
}
}
}
备注:text 用于索引全文值的字段,例如电子邮件正文或产品说明。它们通过分词器传递 ,以在被索引之前将字符串转换为单个术语的列表。分析过程允许Elasticsearch搜索单个单词中 每个完整的文本字段。文本字段不用于排序,很少用于聚合。
keyword 用于索引结构化内容的字段,例如电子邮件地址,主机名,状态代码,邮政编码或标签。它们通常用于过滤,排序,和聚合。keyword字段只能按其确切值进行搜索。
有时候一个字段同时拥有全文类型(text)和关键字类型(keyword)是有用的:一个用于全文搜索,另一个用于聚合和排序。
number_of_shards 是指索引要做多少个分片,只能在创建索引时指定,后期无法修改。
number_of_replicas 是指每个分片有多少个副本,后期可以动态修改
primary shard(主分片):每个文档都存储在一个分片中,当你存储一个文档的时候,系统会首先存储在主分片中,然后会复制到不同的副本中。默认情况下,一个索引有5个主分片。你可以在事先制定分片的数量,当分片一旦建立,分片的数量则不能修改。
replica shard(副本分片):每一个分片有零个或多个副本。副本主要是主分片的复制,可以 增加高可用性,提高性能。
默认情况下,一个主分配有一个副本,但副本的数量可以在后面动态的配置增加。
副本必须部署在不同的节点上,不能部署在和主分片相同的节点上。
新增索引数据
PUT /project_v1/_doc/1
{
"name_en":"encourage",
"name_cn":"营销码",
"project_type":"营销",
"people_count":4,
"order_count":1000000,
"date":"2019-04-01"
}
查询索引数据
GET /project_v1/_search
{
"query": {
"match_all": {}
}
}
匹配查询 match
GET /project_v1/_search
{
"query": {
"match": {
"name_cn": "营销"
}
}
}
过滤查询 Filter
GET /project_v1/_search
{
"query": {
"bool": {
"filter": {
"range": {
"date": {
"gte": "2020-04-01"
}
}
}
}
}
}
六、Elasticsearch 聚合查询
1.聚合的概念
官方对聚合有四个关键字:Metric(指标)、Bucketing(桶)、Pipeline(管道)、Matrix(矩阵),在查询请求体中以aggregations语法来定义聚合分析,也可简写成aggs
Metric(指标):指标分析类型,如计算最大值、最小值、平均值等(对桶内的文档进行聚合分析的操作)
Bucket(桶):分桶类型,类似sql中的group by语法(满足特定条件的文档的集合)
Pipeline(管道):管道分析类型,基于上一级的聚合分析结果进行再分析
Matrix(矩阵):矩阵分析类型(聚合是一种面向数值型的聚合,用于计算一组文档字段中的统计信息)
2.Metric(指标)聚合
Metric聚合分析分为单值分析和多值分析两类
1、单值分析,只输出一个分析结果
关键字有min, max,avg,sum,cardinality
2、多值分析,输出多个分析结果
关键字有stats,extended_stats,percentile_rank,top hits
1)、min, max,avg, sum
GET /project_v1/_search
{
"size": 0,
"aggs": {
"min_people_count": {
"min": {
"field": "people_count"
}
},
"max_order_count":{
"max": {
"field": "order_count"
}
},
"avg_order_count":{
"avg": {
"field": "order_count"
}
},
"sum_order_count":{
"sum": {
"field": "order_count"
}
}
}
}
2)、cardinality
cardinality 关键字: 求唯一值,即不重复的字段有多少(相当于sql中的distinct)
GET /project_v1/_search
{
"size": 0,
"aggs": {
"cardinality_project_type": {
"cardinality": {
"field": "project_type"
}
}
}
}
3)、stats
统计信息,请求后会直接显示各种聚合结果(count,min,max,avg,sum)
GET /project_v1/_search
{
"size": 0,
"aggs": {
"stats_order_count": {
"stats": {
"field": "order_count"
}
}
}
}
4)、extended_stats
扩展的统计信息,比stats返回更多的统计信息
GET /project_v1/_search
{
"size": 0,
"aggs": {
"extended_stats_order_count": {
"extended_stats": {
"field": "order_count"
}
}
}
}
3.Bucket(桶)聚合
Bucket可以理解为一个桶,它会遍历文档中的内容,凡是符合某一要求的就放在一个桶中,分桶相当于sql中的group by
关键字有Terms Aggregation,Filter Aggregation,Histogram Aggregation,Date Aggregation
1)、Terms Aggregation
根据某一项的每个唯一的值来聚合
GET /project_v1/_search
{
"size": 0,
"aggs": {
"terms_project_type": {
"terms": {
"field": "project_type",
"size": 3
}
}
}
}
2)、Filter Aggregation
指具体的域和具体的值,可以在Terms Aggregation 的基础上进行了过滤,只对特定的值进行了聚合
#查营销类型的总订单数
GET /project_v1/_search
{
"size": 0,
"aggs": {
"filter_project_type": {
"filter": {
"term": {
"project_type": "营销"
}
},
"aggs": {
"sum_order_count": {
"sum": {
"field": "order_count"
}
}
}
}
}
}
3)、Histogram Aggregation
Histogram与Terms聚合类似,都是数据分组,区别是Terms是按照Field的值分组,而Histogram可以按照指定的间隔对Field进行分组
#项目规模
GET /project_v1/_search
{
"size": 0,
"aggs": {
"project_scale": {
"histogram": {
"field": "people_count",
"interval": 1
}
}
}
}
4)、Date Aggregation
针对时间格式数据的直方图聚合,基本特性与Histogram Aggregation一致
#项目发展史
GET /project_v1/_search
{
"size": 0,
"aggs": {
"date_by_day": {
"date_histogram": {
"field": "date",
"calendar_interval": "day",
"min_doc_count": 1
}
}
}
}
4.Pipeline(管道)聚合
管道的概念:支持对聚合分析的结果,再次进行聚合分析
#查项目类型最少人数的项目类型
GET /project_v1/_search
{
"size":0,
"aggs":{
"project_type":{
"terms": {
"field": "project_type",
"size": 3
},
"aggs":{
"sum_people_count":{
"sum": {
"field": "people_count"
}
}
}
},
"min_people_count_by_project_type":{
"min_bucket": {
"buckets_path": "project_type>sum_people_count"
}
}
}
}
5.总结
Metric(指标):分类并对一组文档进行sum、avg等数学运算
Bucketing(桶):桶聚合,常规的分类然后计算每个分类的文档数量
Pipeline(管道):对聚合的结果再次聚合
Matrix(矩阵):可在多个字段上计算,生成矩阵结果
七,通过SQL查询Elasticsearch
1.为什么用SQL查询
Elasticsearch 的官方查询语言是 Query DSL,既然是官方指定的,说明最吻合 ES 的强大功能,为ES做支撑。
其实,SQL 作为一个数据库查询语言,它语法简洁,书写方便而且大部分服务端程序员都清楚了解和熟知它的写法。但是作为一个 ES 新人来说,就算他已经是一位编程界的老江湖,但是如果他不熟悉 ES ,那么他如果要使用公司已经搭好的 ES 服务,他必须要先学习 Query DSL,学习成本也是一项影响技术开发进度的因素而且不稳定性高。但是如果 ES 查询支持 SQL的话,那么也许就算他是工作一两年的同学,他虽然不懂 ES的复杂概念,他也能很好的使用 ES 而且顺利的参加到开发的队伍中,毕竟SQL 都会写
2.Elasticsearch-SQL
Elasticsearch-SQL不属于 Elasticsearch 官方的,它是 NLPChina(中国自然语言处理开源组织)开源的一个 ES 插件,主要功能是通过 SQL 来查询 ES,其实它的底层是通过解释 SQL,将SQL 转换为 DSL 语法,再通过DSL 查询。
查询语法
SELECT fields from indexName/type WHERE conditions
表名 tableName 的地方现在改为了索引名 indexName,如果有索引Type ,则indexName/type
POST _sql?format=txt
{
"query": "select * from project_index limit 10"
}
SQL翻译成DSL语句
POST _sql/translate
{
"query": "select name_en,COUNT(*) from project_index GROUP BY name_en"
}
八、注意点
1.版本问题
es 5到7的版本变动很大,其中包括type的变动
- 5.x 支持多种type
- 6.x 只能有一种type
- 7.x 将去除type 没有类型的概念了
2.ES并不可靠
ES不是可靠的存储系统,不是数据库,它有丢数据的风险。ES不是实时系统,数据写入成功只是trans log成功(类似于mysql的bin log),写入成功后立刻查询查不到是正常的。因为数据此刻可能还在内存里而不是进入存储引擎里。
3.同步问题
在需要添加新数据与新字段的时候,如果elasticSearch进行搜索是可能需要重新修改格式。之前的数据需要重新同步,对数据的管理有很多困难。
九、SpringBoot集成Elasticsearch
1.引入依赖
<properties>
<!--告诉springboot我们处理的ES的版本-->
<elasticsearch.version>7.10.2</elasticsearch.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-elasticsearch</artifactId>
</dependency>
</dependencies>
2.yml配置es集群
spring:
elasticsearch:
rest:
uris:
- 192.168.53.112:9200
- 192.168.53.113:9200
- 192.168.53.114:9200
3.简单Test
3.1 创建索引以及分片设置
@Test
public void createIndex() throws Exception{
//1 创建索引并设置分片
//1.1 创建一个RestHightLevelClient对象,相当于和服务端建立连接。
RestHighLevelClient client = new RestHighLevelClient(RestClient.builder(
//没有集群的话 此处可new 一个即可。
new HttpHost("192.168.53.112",9200)
new HttpHost("192.168.53.113",9200),
new HttpHost("192.168.53.114",9200),
));
//1.2 使用client的索引管理的对象,indices()返回索引管理对象。
IndicesClient indicesClient = client.indices();
//两个参数
//1.2.1 创建索引请求对象 参数:创建的索引库的名称
CreateIndexRequest request = new CreateIndexRequest("hello")
.settings(Settings.builder()
.put("number_of_shards", 5)
.put("number_of_replicas", 1)
.build()
);
//1.2.2 请求选项,使用默认值。配置请求头,主要用于认证。
CreateIndexResponse response = indicesClient.create(request, RequestOptions.DEFAULT);
//显示结果
System.out.println(response.toString());
}
3.2 创建索引库并设置mapping信息
@Test
public void createIndexAndMapping() throws Exception{
RestHighLevelClient client = new RestHighLevelClient(RestClient.builder(
//没有集群的话 此处可new 一个即可。
new HttpHost("192.168.53.112",9200)
new HttpHost("192.168.53.113",9200),
new HttpHost("192.168.53.114",9200),
));
//创建json数据
XContentBuilder mappings = XContentFactory.jsonBuilder()
.startObject()
.startObject("properties")
.startObject("id")
.field("type","long")
.endObject()
.startObject("title")
.field("type","text")
.field("analyzer","ik_smart")
.field("store",true)
.endObject()
.endObject()
.endObject();
//创建索引请求对象 参数:创建的索引库的名称,分片副片数量以及mapping信息
CreateIndexRequest request = new CreateIndexRequest("hello1")
.settings(Settings.builder()
.put("number_of_shards", 5)
.put("number_of_replicas", 1)
.build()
)
.mapping(mappings);
//两个参数
//1 创建索引请求对象 参数:创建的索引库的名称
//2 请求选项,使用默认值。配置请求头,主要用于认证。
CreateIndexResponse response = client.indices().create(request, RequestOptions.DEFAULT);
//显示结果
System.out.println(response.toString());
}
}
3.3删除索引库
@Test
public void deleteIndex() throws Exception{
RestHighLevelClient client = new RestHighLevelClient(RestClient.builder(
//没有集群的话 此处可new 一个即可。
new HttpHost("192.168.53.112",9200)
new HttpHost("192.168.53.113",9200),
new HttpHost("192.168.53.114",9200),
));
//删除索引库
AcknowledgedResponse response = client.indices().delete(new DeleteIndexRequest("hello"), RequestOptions.DEFAULT);
//显示结果
System.out.println(response.toString());
}
3.4 添加索引库字段信息
@Test
public void putIndex() throws Exception{
RestHighLevelClient client = new RestHighLevelClient(RestClient.builder(
//没有集群的话 此处可new 一个即可。
new HttpHost("192.168.53.112",9200)
new HttpHost("192.168.53.113",9200),
new HttpHost("192.168.53.114",9200),
));
String mappings = "{n" +
"ttt"properties":{n" +
"tttt"id":{n" +
"ttttt"type" : "long"n" +
"tttt},n" +
"tttt"title" :{n" +
"ttttt"type" : "text",n" +
"ttttt"analyzer" : "ik_smart",n" +
"ttttt"store" : truen" +
"tttt},n" +
"tttt" content" :{n" +
"ttttt"type" : "text",n" +
"ttttt"analyzer" : "ik_smart",n" +
"ttttt"store" :truen" +
"tttt}n" +
"ttt}n" +
"tt}";
//将字符串以json形式发送
PutMappingRequest request = new PutMappingRequest("hello1")
.source(mappings, XContentType.JSON);
//修改索引库
AcknowledgedResponse response = client.indices().putMapping(request, RequestOptions.DEFAULT);
//显示结果
System.out.println(response.toString());
}
十、Elasticsearch文档管理
0.抽取ES连接对象的公共方法
//原生客户端类,即ESjava客户端。
private RestHighLevelClient client;
public void init(){
//1.1 创建一个RestHightLevelClient对象,相当于和服务端建立连接。
client = new RestHighLevelClient(RestClient.builder(
//没有集群的话 此处可new 一个即可。
new HttpHost("192.168.53.112",9200)
new HttpHost("192.168.53.113",9200),
new HttpHost("192.168.53.114",9200),
));
}
1.添加文档
使用RestHightLevelClient对象。
使用client对象的index方法添加文档
创建IndexRequest对象,其中包含了索引库名称、文档id、文档的内容
{“id”:“1”,“title”:“测试文档1”,“content”:“测试文档中的内容”}
public void addDocument() throws Exception{
String document = "{"id":1, "title":"这是测试文章", "content":"xxxxx"}";
//创建IndexRequest对象,其中包含索引库名称,文档id,文档内容
IndexRequest request = new IndexRequest()
.index("hello1")
.id("1")
.source(document, XContentType.JSON);
IndexResponse response = client.index(request, RequestOptions.DEFAULT);
System.out.println(response.toString());
}
2.更新文档
使用client对象的update方法。
需要UpdateRequest参数:
1.更新的索引
2.更新的文档的id
3.更新的文档内容
public void updateDocument() throws Exception{
String document = "{"id":1, "title":"这是测试文章更细的", "content":"new update"}";
//创建IndexRequest对象,其中包含索引库名称,文档id,文档内容
UpdateRequest request = new UpdateRequest()
.index("hello1")
.id("1")
.doc(document, XContentType.JSON);
UpdateResponse response = client.update(request, RequestOptions.DEFAULT);
System.out.println(response.toString());
}
3.删除文档
使用client的delete方法
需要DeleteRequest对象,需要两个参数
1.操作的索引
2.文档的id
public void deleteDocument() throws Exception{
//创建IndexRequest对象,其中包含索引库名称,文档id,文档内容
DeleteRequest request = new DeleteRequest("hello1", "1");
DeleteResponse response = client.delete(request, RequestOptions.DEFAULT);
System.out.println(response.toString());
}
4.根据id查询文档
使用client对象的get方法。
需要使用GetRequest对象,两个参数:
1.操作的索引
2.文档的id
public void getDocument() throws Exception{
//创建IndexRequest对象,其中包含索引库名称,文档id,文档内容
GetRequest request = new GetRequest("hello1", "1");
GetResponse response = client.get(request, RequestOptions.DEFAULT);
System.out.println(response.toString());
}
5.批量查询文档
使用client对象的bulk方法。
BulkRequest对象,使用add方法,添加要批量处理的请求。
支持的处理:IndexRequest,DeleteRequest,UpdateRequest
public void bulkDocument() throws Exception{
//json数据
String jsonData = "[" +
"{"id":3, "title":"这是测试文章1", "content":"xxxxx", "comment":"备注信息", "mobile":"13344556677"}n" +
"{"id":4, "title":"这是一篇文章2", "content":"xxxxx", "comment":"备注信息", "mobile":"13344556677"}n" +
"{"id":5, "title":"这是一篇文章3", "content":"xxxxx", "comment":"备注信息", "mobile":"13344556677"}]";
//转换成json格式字符串
JSONArray jsonArray = JSONObject.parseArray(jsonData);
//创建IndexRequest对象,其中包含索引库名称,文档id,文档内容
BulkRequest request = new BulkRequest();
jsonArray.stream()
.forEach(json -> {
IndexRequest r = new IndexRequest()
.index("hello1")
.id(((JSONObject) json).getString("id"))
.source(((JSONObject) json).toJSONString(), XContentType.JSON);
request.add(r);
});
BulkResponse response = client.bulk(request, RequestOptions.DEFAULT);
System.out.println(response.toString());
}
十一、ElasticsearchRestTemplate类与ElasticsearchRepository类
SpringData对ES的封装ElasticsearchRestTemplate类,可直接使用,此类在ElasticsearchRestTemplate基础上进行性一定程度的封装,使用起来更方便灵活,拓展性更强。
ElasticsearchRepository可以被继承操作ES,是SpringBoot对ES的高度封装,操作最为方便,但牺牲了灵活性。
索引库实体类
@Data
@Document(indexName = "blog_1", shards = 5, replicas = 1)
public class Blog {
@Field(type = FieldType.Long, store = true)
private Long id;
//type = FieldType.Text 字段类型为text
//analyzer = "ik_max_word" 分词器为"ik_max_word"
//store = true 存储 => 是
@Field(type = FieldType.Text, analyzer = "ik_max_word", store = true)
private String title;
@Field(type = FieldType.Text, analyzer = "ik_max_word", store = true)
private String content;
@Field(type = FieldType.Text, analyzer = "ik_max_word", store = true)
private String comment;
@Field(type = FieldType.Keyword, store = true)
private String mobile;
}
1、使用ElasticsearchRestTemplate类
a)、创建索引库
@Autowired
private ElasticsearchRestTemplate template;
/**
* 创建索引库
*/
public void createIndex(){
//创建索引库
template. indexOps(IndexCoordinates.of("mytest")).create();
}
b)、创建索引库并实体类设置mapping
1)创建索引库
template.indexOps(IndexCoordinates.of(“mytest”)).create();
2)设置mapping信息
需要创建一个实体类,其中配置实体类和文档的映射关系,使用注解配置。
可以从Entity中生成mapping信息。
public void putMapping(){
//创建索引库
Document mapping = template.indexOps(IndexCoordinates.of("mytest")).createMapping(Blog.class);
template.indexOps(IndexCoordinates.of("mytest")).putMapping(mapping);
}
c)、删除索引库
//删除索引库
public void deleteIndex(){
template.indexOps(IndexCoordinates.of("hello1")).delete();
}
d)、索引库查询
public void maxQueryTest(){
NativeSearchQuery builder = new NativeSearchQueryBuilder()
//多字段查询 (高亮跟查询条件有关)
.withQuery(QueryBuilders.multiMatchQuery("8", "id","title"))
//增加过滤条件, 可以设置多个
.withFilter(QueryBuilders.boolQuery()
//增加bool查询:should的term关键字查询
.should(QueryBuilders.termQuery("title", "文章"))
.should(QueryBuilders.termQuery("content","xxx"))
)
//增加过滤条件的关键字查询
.withFilter(QueryBuilders.termQuery("mobile", "13344556677"))
//分页设置
.withPageable(PageRequest.of(0,5))
//设置高亮
.withHighlightBuilder(new HighlightBuilder()
//高亮显示的字段
.field("title")
//高亮显示的字段
.field("content")
//高亮显示的前缀
.preTags("<em>")
//高亮显示的后缀
.postTags("</em>")
)
//添加聚合查询
.addAggregation(new TermsAggregationBuilder("mobile_group").field("mobile"))
.build();
//基于Blog.class 类型返回的结果
SearchHits<Blog> searchHits = template.search(builder, Blog.class);
//从searchHits取相关数据
long totalHits = searchHits.getTotalHits(); //取总记录数
List<SearchHit<Blog>> list = searchHits.getSearchHits(); //取每条数据放入集合中
System.out.println("总记录数为:" + totalHits);
list.forEach(blogSearchHit -> {
//取原生文档对象
Blog blog = blogSearchHit.getContent();
System.out.println(blog);
//取高亮对象
Map<String, List<String>> highlightFields = blogSearchHit.getHighlightFields();
System.out.println(highlightFields);
//取高亮对象 放到Blog里去 这样就将Blog和高亮结合输出了
String title = highlightFields.get("title").get(0);
//String content = highlightFields.get("content").get(0);
blog.setTitle(title);
//blog.setContent(content);
System.out.println(blog);
});
//取聚合结果
Aggregations aggregations = searchHits.getAggregations();
System.out.println(aggregations.toString());
}
2、使用ElasticsearchRepository类
a)、创建接口继承ElasticsearchRepository
public interface BlogRepository extends ElasticsearchRepository<Blog, Long> {
/**
* 定义一个方法查询:根据title查询es
*
* 原因: ElasticsearchRepository会分析方法名,参数对应es中的field(这就是灵活之处)
* @param title
* @return java.util.List<com.yt.cubemall.search.model.Blog>
*/
List<Blog> findByTitle(String title);
/**
* 定义一个方法查询: 根据title,content查询es
*/
List<Blog> findByTitleAndContent(String title, String content);
}
b)、使用BlogRepository接口
public class BlogRepositoryTest {
@Autowired
private BlogRepository blogRepository;
/**
* 添加文档
*/
@Test
public void addDocument(){
Blog blog = new Blog();
for (int i = 0; i < 10; i++) {
blog.setId((long)i+1);
blog.setTitle("测试spring集成es"+i+1);
blog.setContent("sjihfapf"+i+1);
blog.setComment("注释内容"+i+1);
blog.setMobile("12345678901");
blogRepository.save(blog);
}
}
/**
* 更新文档
*/
@Test
public void updateDocument(){
Optional<Blog> optional = blogRepository.findById(1l);
if (optional.isPresent()){
Blog blog = optional.get();
blog.setTitle("hello update");
blogRepository.save(blog);
}
}
/**
* 删除文档
*/
@Test
public void deleteDocument() {
blogRepository.deleteById(1l);
}
/**
* 查询所有 文档
*/
@Test
public void getDocument() {
//根据id查找
//Optional<Blog> optional = blogRepository.findById(1l);
//Blog blog = optional.get();
//System.out.println(blog);
//查找全部
//Iterable<Blog> all = blogRepository.findAll();
//all.forEach(blog -> System.out.println(blog));
//分页查找全部
Iterable<Blog> all = blogRepository.findAll(PageRequest.of(1,10));
all.forEach(blog -> System.out.println(blog));
}
/**
* 自定义方法:根据title内容查询索引库
* /
@Test
public void testFindByTitle(){
List<Blog> blogList = blogRepository.findByTitle("测试");
blogList.stream().forEach(System.out::println);
}
/**
* 自定义方法:根据title,content内容查询索引库
* /
@Test
public void testFindByTitleAndContent(){
List<Blog> blogList = blogRepository.findByTitleAndContent("测试", "sjihfapf");
blogList.stream().forEach(System.out::println);
}
}
}