赞
踩
Elasticsearch 是一个基于 Lucene 的搜索服务器,它提供了一个分布式多用户能力的全文搜索引擎,基于 RESTful web 接口
Elasticsearch 是用 Java 开发的,并作为 Apache 许可条款下的开放源码发布,是当前流行的企业级搜索引擎。设计用于云计算中,能够达到实时搜索、稳定可靠、快速安装、使用方便
优点:
Lucene 不是一个完整的全文检索引擎,而是一个全文检索引擎的架构,提供了完整的查询引擎和索引引擎,部分文本分析引擎。想要使用它,你必须使用 Java 来作为开发语言并将其直接集成到你的应用中,更糟糕的是,Lucene 非常复杂,你需要深入了解检索的相关知识来理解它是如何工作的。
Solr 是 Apache Lucene 项目的开源企业搜索平台。其主要功能包括全文检索、命中标示、分面搜索、动态聚类、数据库集成,以及富文本(如 Word、PDF)的处理,Solr 是高度可扩展的,并提供了分布式搜索和索引复制。Solr 是最流行的企业级搜索引擎,Solr4 还增加了 NoSQL 的支持
当单纯的对已有数据进行搜索时,Solr更快
当实时建立索引时,Solr 会产生IO 阻塞,查询性能较差,Elasticsearch 具有明显的优势
随着数据量的增加,Solr 的搜索效率会变得更低,而 Elasticsearch 却没有明显的变化
综上所述,Solr 的架构并不适合实时搜索的应用
实际生产环境测试,下图为将搜素引擎从Solr转到Elasticsearch以后的平均查询速度有了 50 倍的提升
![在这里插入图片描述]](https://img-blog.csdnimg.cn/a3320ae34bb842629c060e3bcaf92e8e.png)
官网下载地址:https://www.elastic.co/cn/downloads/elasticsearch
当前安装包:https://www.elastic.co/cn/downloads/past-releases/elasticsearch-7-17-2
# 下载
wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-7.17.2-linux-x86_64.tar.gz
# 解压
tar -zxvf elasticsearch-7.17.2-linux-x86_64.tar.gz -C /usr/local/
bin 启动文件目录
config 配置文件目录
log4j2.properties 日志配置文件
jvm.options java虚拟机相关配置文件
elasticsearch.yml Elasticsearch配置文件
lib 相关jar包目录
logs 日志目录
modules 功能模块目录
plugins 插件目录
# 配置es的集群名称,默认是elasticsearch # 如果在同一网段下有多个集群,就可以用这个属性来区分不同的集群。 # 保证每个节点的名称相同,如此就能都处于一个集群之内了 cluster.name: my-application # 节点名称 # 每一个节点的名称,必须不一样 node.name: node-1 # 默认只允许本机访问,修改为0.0.0.0后则可以远程访问 network.host: 0.0.0.0 # 主节点,作用主要是用于来管理整个集群,负责创建或删除索引,管理其他非master节点 # 允许节点是否可以成为一个master节点,ES是默认集群中的第一台机器成为master,如果这台机器停止就会重新选举。(默认开启) node.master: true # 数据节点,用于对文档数据的增删改查 # 允许该节点存储索引数据(默认开启) node.data: true # 默认端口 http.port: 9200 # 集群列表 # 设置集群中master节点的初始列表,可以通过这些节点来自动发现新加入集群的节点 discovery.seed_hosts: ["ip:port", "ip:port", "ip:port"] # 初始化master节点 # 初始化一个新的集群时需要此配置来选举master cluster.initial_master_nodes: ["node-1", "node-2", "node-3"]
Elasticsearch 通过 jvm.options 中的 Xms 和 Xmx 设置堆的大小
应该来讲 Xms 和 Xmx 设为相同的值
Xms 和 Xmx 的值不应超过物理内存的 50%,因为 Elasticsearch 在堆内存之外还需要将内存用于其他。例如 Elasticsearch 需要利用堆外内存来进行网络通信,依赖操作系统的文件系统缓存来有效访问文件,而 JVM 本身也需要一些内存。
# 设置JVM堆的大小为 1G
-Xms1g
-Xmx1g
# 创建用户
useradd es
# 修改用户密码
passwd es
# 创建所属组并授权es目录权限
chown es:es -R /usr/local/elasticsearch/
# 切换到es用户
su es
# 进入Elasticsearch bin目录
cd /usr/local/elasticsearch/bin
# 启动Elasticsearch
./elasticsearch
# 后台启动Elasticsearch
./elasticsearch -d
max file descriptors [4096] for elasticsearch process is too low, increase to at least [65535]
从错误信息看出应该是 Elasticsearch 程序需要的最小 max file descriptors 值是 65536,但是我的host只配置了4096(默认值)
查看 max file descriptors:
[root@192 ~]# ulimit -Hn
4096
[root@192 ~]# ulimit -Sn
1024
ulimit -Hn: 是max number of open file descriptors的hard限制
ulimit -Sn: 是max number of open file descriptors的soft限制
接下来要把这两个值改大,在 /etc/security/limits.conf 中添加如下内容
es hard nofile 65536
es soft nofile 65536
改完需要重新登录才能生效,或者切换用户:
[root@192 ~]# su es
[es@192 root]$ ulimit -Hn
65536
[es@192 root]$ ulimit -Sn
65536
max virtual memory areas vm.max_map_count [65530] is too low, increase to at least [262144]
elasticsearch用户拥有的内存权限太小,至少需要262144,解决办法:
切换root用户,在 /etc/sysctl.conf 文件最后添加如下内容,即可永久修改
vm.max_map_count = 262144
刷新配置文件:
[root@192 ~]# sysctl -p
vm.max_map_count = 262144
再次启动成功:
访问 http://127.0.0.1:9200
jdk 环境报错更改
# 修改jdk环境变量
vim bin//elasticsearch-env
# 首行添加
JAVA_HOME="/usr/local/elasticsearch/jdk"
启动从节点时无法加入到集群,报错信息:
failed to validate incoming join request from node
Caused by: org.elasticsearch.cluster.coordination.CoordinationStateRejectedException: This node previously joined a cluster with UUID [Kzs2RYf-Qe6Y5VWrtmSrQw] and is now trying to join a different cluster with UUID [eRJNKjZjRfCpXJZbavulUQ]. This is forbidden and usually indicates an incorrect discovery or cluster bootstrapping configuration. Note that the cluster UUID persists across restarts and can only be changed by deleting the contents of the node's data paths [] which will also remove any data held by this node.
原因:推测是因为该节点之前启动过ES,已经创建了data文件夹,与要加入的集群冲突。
解决:因为该节点之前已经启动过,有历史数据没有清理,把该节点的data文件夹删了,再次启动就行
Elasticsearch 只是后端提供各种RESTful API ,那么怎么直观的看它的信息呢?
elasticsearch-head是一款专门针对于elasticsearch的客户端工具,用来展示数据。
elasticsearch-head是基于JavaScript语言编写的,可以使用 npm 部署,npm 是 Nodejs 下的包管理器
安装Nodejs环境
下载地址:http://nodejs.cn/download/
当前安装包:https://npmmirror.com/mirrors/node/v16.16.0/node-v16.16.0-linux-x64.tar.xz
# 下载
wget https://npmmirror.com/mirrors/node/v16.16.0/node-v16.16.0-linux-x64.tar.xz
# 解压
tar -xvf node-v16.16.0-linux-x64.tar.xz -C /usr/local/
# 修改解压目录名称
cd /usr/local/
mv node-v16.16.0-linux-x64 node-v16.16.0
# 配置环境变量
vim /etc/profile
# 追加 ":/usr/local/node-v16.16.0/bin" 内容
# 刷新配置
source /etc/profile
验证:
[root@192 local]# node -v
v16.16.0
[root@192 local]# npm -v
8.11.0
Elasticsearch-head 安装
下载地址:https://github.com/mobz/elasticsearch-head.git
当前安装包:https://github.com/mobz/elasticsearch-head/archive/refs/heads/master.zip
# 解压 unzip -d /usr/local/ elasticsearch-head-master.zip # 更改解压目录名称 mv elasticsearch-head-master elasticsearch-head # 安装cnpm的命令 npm install -g cnpm --registry=https://registry.npm.taobao.org cd elasticsearch-head/ # 安装依赖 cnpm install # 启动服务 npm run start #或者 npm run-script start # 后台启动 nohup npm run-script start &
启动服务:npm run start
[root@192 elasticsearch-head]# npm run start
> elasticsearch-head@0.0.0 start
> grunt server
Running "connect:server" (connect) task
Waiting forever...
Started connect web server on http://localhost:9100
访问: http://127.0.0.1:9100/
使用Elasticsearch-head插件访问elasticsearch
连接失败,要允许跨域:
配置es的elasticsearch.yml文件解决跨域问题:
http.cors.enabled: true
http.cors.allow-origin: "*"
重启ES,再次连接:
head我们可以把它当做数据展示工具,后续的查询可以使用 Kibana
了解ELK
Kibana
下载地址:https://www.elastic.co/cn/kibana/
当前安装包:https://artifacts.elastic.co/downloads/kibana/kibana-7.17.2-linux-x86_64.tar.gz
# 下载
wget https://artifacts.elastic.co/downloads/kibana/kibana-7.17.2-linux-x86_64.tar.gz
# 解压
tar -zxvf kibana-7.17.2-linux-x86_64.tar.gz -C /usr/local/
# 解压后重命名
mv kibana-7.17.2-linux-x86_64 kibana-7.17.2
# 启动服务
cd /usr/local/kibana-7.17.2/bin
./kibana
配置文件:config/kibana.yml
# kibana默认端口为5601 server.port: 5601 # kibana服务器地址 server.host: "0.0.0.0" # es服务器地址 elasticsearch.hosts: ["http://ip:port", "http://ip:port", "http://ip:port"] # 创建的索引名字 kibana.index: ".kibana" # 设置kibana日志存放路径 logging.dest: stdout # 设置使用中文显示页面 i18n.locale: "zh-CN"
启动 Kibana
kibana 不支持root用户启动,如果是要用root用户启动,就在后面加 --allow-root ,要么就切换用户执行
# 配置目录权限给es用户
chown es:es -R /usr/local/kibana-7.17.2/
# 切换es用户
su es
# 启动服务
cd /usr/local/kibana-7.17.2/bin
./kibana
# 后台启动
nohup ./kibana &
访问: http://192.168.10.201:5601/
控制台:
最近更新到Elastic Stack 7.13以上版本的朋友可能注意到了,在默认不开启Elastic 安全功能时,Kibana的搜索结果页面会多出一行提示,建议我们开启ElasticSearch 安全功能。
在个人学习或者内网开放ES+VPN连接的情况下我们完全不需要开启安全功能,其他情况在生产集群中还是建议开启安全选项的。
这是因为没有显式禁用安全选项导致的,也就是说ElasticSearch会提示你是不是忘了启用这个选项,只要在配置文件中显式禁用即可取消这个提示。
在elasticsearch.yml
配置禁用安全选项xpack.security.enabled
,之后重启ElasticSearch即可:
xpack.security.enabled: false
Lucene 作为 Apache 开源的一款搜索工具,一直以来是实现搜索功能的神兵利器,现今火热的 Solr 和 Elasticsearch 均基于该工具包进行开发,而 Lucene 之所以能在搜索中发挥至关重要的作用正是因为倒排索引
。
搜索的核心需求是全文检索,全文检索简单来说就是要在大量文档中找到包含某个单词出现的位置,在传统关系型数据库中,数据检索只能通过 like 来实现
这种实现方式实际会存在很多问题:
搜索的核心目标实际上是保证搜索的效果和性能,为了高效的实现全文检索,我们可以通过倒排索引来解决
倒排索引是区别于正排索引的概念:
forward index
是以文档对象的唯一 ID 作为索引,以文档内容作为记录的结构。inverted index
指的是将文档内容中的单词作为索引,将包含该词的文档 ID 作为记录的结构。通俗来讲,正向索引是通过 key 找 value,反向索引则是通过 value 找 key
正向索引(forward index)
当用户发起查询时(假设查询为一个关键词),搜索引擎会扫描索引库中的所有文档,找出所有包含关键词的文档,这样依次从文档中去查找是否含有关键词的方法叫做正向索引。互联网上存在的网页(或称文档)不计其数,这样遍历的索引结构效率低下,无法满足用户需求。
以文档的ID为关键字,表中记录文档中每个字的位置信息,查找时扫描表中每个文档中字的信息,直到找出所有包含查询关键字的文档
这种组织方法在建立索引的时候结构比较简单,建立比较方便且易于维护
优缺点:
反向索引(inverted index),一般也被别人称之为倒排索引
为了增加效率,搜索引擎会把正向索引变为反向索引,即把 “文档→单词” 的形式变为 “单词→文档” 的形式
倒排索引主要由 单词词典(Term Dictionary)和 倒排列表(Posting List)组成
倒排列表
单词字段和倒排列表整合在一起的结构如下:
搜索引擎需要处理的文档集合往往都是动态集合,即在建好初始的索引后,不断有新文档进入系统,同时原先的文档集合内有些文档可能被删除或更改
动态索引通过在内存中维护临时索引,可以实现对动态文档和实时搜索的支持
服务器内存总是有限的,随着新加入系统的文档越来越多,临时索引消耗的内存也会随之增加
当最初分配的内存将被使用完时,要考虑将临时索引的内容更新到磁盘索引中,以释放内存空间来容纳后续的新进文档
索引基本更新思想:
常见的索引更新策略主要有四种:完全重建策略、再合并策略、原地更新策略及混合策略
完全重建策略
当新增文档到达一定数量,将新增文档和原先的老文档整合,然后利用静态索引创建方法对所有文档重建索引,新索引建立完成后老索引会被遗弃。
此法代价高,但是主流商业搜索引擎一般是采用此方式来维护索引的更新(这句话是书中原话)
优缺点:
再合并策略
当新增文档进入系统,解析文档,之后更新内存中维护的临时索引,文档中出现的每个单词,在其倒排表列表末尾追加倒排表列表项;一旦临时索引将指定内存消耗光,即进行一次索引合并,这里需要倒排文件里的倒排列表存放顺序已经按照索引单词字典顺序由低到高排序,这样直接顺序扫描合并即可。
优缺点:
原地更新策略
混合策略
分析器(Analyzer)都由三种构件块组成:character filters、tokenizers、token filters
IK Analyzer 插件将 Lucene IK分析器集成到 Elasticsearch 中,支持自定义词典
中文的分词器现在大家比较推荐的就是 IK 分词器,当然也有一些其他的分词器,比如 SmartCN、HanLP
IK 分词器的粒度:
下载地址:https://github.com/medcl/elasticsearch-analysis-ik
当前安装包:https://github.com/medcl/elasticsearch-analysis-ik/releases/download/v7.17.2/elasticsearch-analysis-ik-7.17.2.zip
# 在es plugins 目录下新建目录 analysis-ik mkdir analysis-ik [root@192 plugins]# ll 总用量 0 drwxr-xr-x. 2 root root 6 8月 3 12:10 analysis-ik # 解压到指定目录 unzip elasticsearch-analysis-ik-7.17.2.zip -d /usr/local/elasticsearch/plugins/analysis-ik/ [root@192 plugins]# cd analysis-ik/ [root@192 analysis-ik]# ll 总用量 1432 -rw-r--r--. 1 root root 263965 1月 18 2022 commons-codec-1.9.jar -rw-r--r--. 1 root root 61829 1月 18 2022 commons-logging-1.2.jar drwxr-xr-x. 2 root root 4096 1月 18 2022 config -rw-r--r--. 1 root root 54931 4月 1 15:43 elasticsearch-analysis-ik-7.17.2.jar -rw-r--r--. 1 root root 736658 1月 18 2022 httpclient-4.5.2.jar -rw-r--r--. 1 root root 326724 1月 18 2022 httpcore-4.4.4.jar -rw-r--r--. 1 root root 1807 4月 1 15:43 plugin-descriptor.properties -rw-r--r--. 1 root root 125 4月 1 15:43 plugin-security.policy # 授权 chown es:es -R /usr/local/elasticsearch/ [root@192 analysis-ik]# ll 总用量 1432 -rw-r--r--. 1 es es 263965 1月 18 2022 commons-codec-1.9.jar -rw-r--r--. 1 es es 61829 1月 18 2022 commons-logging-1.2.jar drwxr-xr-x. 2 es es 4096 1月 18 2022 config -rw-r--r--. 1 es es 54931 4月 1 15:43 elasticsearch-analysis-ik-7.17.2.jar -rw-r--r--. 1 es es 736658 1月 18 2022 httpclient-4.5.2.jar -rw-r--r--. 1 es es 326724 1月 18 2022 httpcore-4.4.4.jar -rw-r--r--. 1 es es 1807 4月 1 15:43 plugin-descriptor.properties -rw-r--r--. 1 es es 125 4月 1 15:43 plugin-security.policy # 查看插件列表 [root@192 bin]# ./elasticsearch-plugin list analysis-ik
重启ES,进行分词测试:
GET _analyze
{
"analyzer": "ik_max_word",
"text": "今天天气真好"
}
GET _analyze
{
"analyzer": "ik_smart",
"text": "今天天气真好"
}
MySQL 与 Elasticsearch 对比:
MySQL | Elasticsearch |
---|---|
数据库 database | 索引 index |
表 tables | 类型 types |
行记录 rows | 文档 document |
列字段 columns | 字段 field |
表结构定义 schema | 字段定义 mapping |
Elasticsearch是面向文档的,文档是所有可搜索数据的最小基础信息单元。
一个Document就像数据库中的一行记录,文档会被序列化成JSON格式,保持在Elasticsearch中,多个Document存储于一个索引(Index)中。文档以JSON(Javascript Object Notation)格式来表示,而JSON是一个到处存在的互联网数据交互格式。
每一个文档都有一个UniqueID
document核心元数据(元数据:用于标注稳定的相关信息):
_index
代表一个document存放在哪个index中,项目约定结构类似的数据放在一个索引,不同数据放不同索引里,所以同一个index中document结构基本是类似的,个别document多一个或少一个field,这样Elasticsearch对磁盘存储的利用率最高。
每个index有自己独立的shard存储文件,与其他index互不影响。
命名规范:名称小写,不能以 ‘_’, ‘-’, 或 ‘+’ 开头
_type
_id
_version
found
_source
Document 中的字段,是文档中的某一个属性
其中,在 ES 7.x 有两种字符串类型:text 和 keyword,在 ES 5.x 之后 string 类型已经不再支持了。
JSON 文档中同样存在布尔类型,不过 JSON 字符串类型也可以被 ES 转换为布尔类型存储,前提是字符串的取值为 true 或者 false,布尔类型常用于检索中的过滤条件。
二进制类型 binary 接受 BASE64 编码的字符串,默认 store 属性为 false,并且不可以被搜索。
范围类型可以用来表达一个数据的区间,可以分为5种:
integer_range、float_range、long_range、double_range 以及 date_range
复合类型主要有对象类型(object)和嵌套类型(nested)
JSON 字符串允许嵌套对象,一个文档可以嵌套多个、多层对象。可以通过对象类型来存储二级文档,不过由于 Lucene 并没有内部对象的概念,ES 会将原 JSON 文档扁平化,列如文档:
实际上 ES 会将其转换为以下格式,并通过 Lucene 存储,即使 name 是 object 类型
嵌套类型可以看成是一个特殊的对象类型,可以让对象数组独立检索,例如文档
username 字段是一个 JSON 数组,并且每个数组对象都是一个 JSON 对象。如果将 username 设置为对象类型,那么 ES 会将其转换为:
可以看出转换后的 JSON 文档中 first 和 last 的关联丢失了,如果尝试搜索 first 为 wu,last 为 xy 的文档,那么成功会检索出上述文档,但是 wu 和 xy 在原 JSON 文档中并不属于同一个 JSON 对象,应当是不匹配的,即检索不出任何结果。
地理类型字段分为两种:经纬度类型和地理区域类型
经纬度类型字段(geo_point)可以存储经纬度相关信息,通过地理类型的字段,可以用来实现诸如查找在指定地理区域内相关的文档、根据距离排序、根据地理位置修改评分规则等需求。
经纬度类型可以表达一个点,而 geo_shape 类型可以表达一块地理区域,区域的形状可以是任意多边形,也可以是点、线、面、多点、多线、多面等几何类型
特殊类型包括 IP 类型、过滤器类型、Join 类型、别名类型等
IP 类型的字段可以用来存储 IPv4 或者 IPv6 地址,如果需要存储 IP 类型的字段,需要手动定义映射:
提供自动输入关联完成功能,如常见的Baidu搜索框
用于计算字符串token的长度,使用时需提供"analyzer"定义
定义为 percolate 类型的字段会被ES分析为一个查询并保存下,并可用在后续对文档的查询中。Percolate 可以理解为一个预置的查询
定义一个已存在域的别名
Join 类型是 ES 6.x 引入的类型,以取代淘汰的 _parent 元字段,用来实现文档的一对一、一对多的关系,主要用来做父子查询。
Join 类型的 Mapping 如下:
其中,my_join_field 为 Join 类型字段的名称;relations 指定关系:question 是 answer 的父类。
例如定义一个 ID 为 1 的父文档:
接下来定义一个子文档,该文档指定了父文档 ID 为 1:
analyzer 可指定文本和检索所用的分词器
index 可用于设置字段是否被索引,默认为 true,false 即为不可搜索
需要对 Null 值实现搜索使用,只有 keyword 类型才支持设定 null_value
Dynamic Mapping 机制使我们不需要手动定义 Mapping,ES 会自动根据文档信息来判断字段合适的类型,但是有时候也会推算的不对,比如地理位置信息有可能会判断为 Text,当类型如果设置不对时,会导致一些功能无法正常工作,比如 Range 查询。
ES 类型的自动识别是基于 JSON 的格式,如果输入的是 JSON 是字符串且格式为日期格式,ES 会自动设置成 Date 类型;当输入的字符串是数字的时候,ES 默认会当成字符串来处理,可以通过设置来转换成合适的类型;如果输入的是 Text 字段的时候,ES 会自动增加 keyword 子字段,还有一些自动识别如下图所示:
动态映射时 Elasticsearch 的一个重要特性:
万一我想修改 Mapping 的字段类型,能否更改呢?让我们分以下两种情况来探究下:
控制Dynamic Mappings:
“true” | “false” | “strict” | |
---|---|---|---|
文档可索引 | YES | YES | NO |
字段可索引 | YES | NO | NO |
Mapping被更新 | YES | NO | NO |
第一种为ES默认选项,以下两个值都为true,该节点既有成为master的资格还要存储数据,如果该节点被选为了真正的master,还要存储数据,那么该节点的压力相对来说就较大了,生产中不建议这样,即成为了主节点还成为了数据节点。
node.master: true
node.data: true
第二种为主节点模式,该节点只有成为master节点的资格,并不正真存储数据,在没有成为master的情况下还可以进行集群内的请求转发,数据合并等功能,此选择在生产中为master节点。
node.master: true
node.data: false
第三种为数据节点,该节点只存储数据,没有资格成为master,只做数据存储功能;在集群中需要单独设置几个这样的节点,用来存储数据。
node.master: false
node.data: true
第四种为client节点,该节点既没有成为master的资格,还不存储数据,主要是针对海量请求的时候可以进行负载均衡、数据合并、数据查询、请求转发等功能。
node.master: false
node.data: false
# 配置es的集群名称,默认是elasticsearch # 如果在同一网段下有多个集群,就可以用这个属性来区分不同的集群。 # 保证每个节点的名称相同,如此就能都处于一个集群之内了 cluster.name: my-application # 节点名称 # 每一个节点的名称,必须不一样 node.name: node-1 # 主节点,作用主要是用于来管理整个集群,负责创建或删除索引,管理其他非master节点 # 允许节点是否可以成为一个master节点,ES是默认集群中的第一台机器成为master,如果这台机器停止就会重新选举。(默认开启) node.master: true # 数据节点,用于对文档数据的增删改查 # 允许该节点存储索引数据(默认开启) node.data: true # 集群列表 # 设置集群中master节点的初始列表,可以通过这些节点来自动发现新加入集群的节点 discovery.seed_hosts: ["ip:port", "ip:port", "ip:port"] # 初始化master节点 # 初始化一个新的集群时需要此配置来选举master cluster.initial_master_nodes: ["node-1", "node-2", "node-3"]
注意事项:集群需要开启9300端口进行通讯
查看集群状态:http://ip:9200/_cluster/health?pretty # ?pretty JSON格式显示,易读
输出里最重要的就是 status 这行。很多开源的 ES 监控脚本,其实就是拿这行数据做报警判断。status 有三个可能的值:
green/yellow/red 状态是一个概览你的集群并了解眼下正在发生什么的好办法。剩下来的指标给你列出来集群的状态概要:
number_of_nodes 和 number_of_data_nodes 这个命名完全是自描述的。
active_primary_shards 指出你集群中的主分片数量。这是涵盖了所有索引的汇总值。
active_shards 是涵盖了所有索引的所有分片的汇总值,即包括副本分片。
relocating_shards 显示当前正在从一个节点迁往其他节点的分片的数量。通常来说应该是 0,不过在 Elasticsearch 发现集群不太均衡时,该值会上涨。比如说:添加了一个新节点,或者下线了一个节点。
initializing_shards 是刚刚创建的分片的个数。比如,当你刚创建第一个索引,分片都会短暂的处于 initializing 状态。这通常会是一个临时事件,分片不应该长期停留在 initializing 状态。你还可能在节点刚重启的时候看到 initializing 分片:当分片从磁盘上加载后,它们会从 initializing 状态开始。
unassigned_shards 是已经在集群状态中存在的分片,但是实际在集群里又找不着。通常未分配分片的来源是未分配的副本。比如,一个有 5 分片和 1 副本的索引,在单节点集群上,就会有 5 个未分配副本分片。如果你的集群是 red 状态,也会长期保有未分配分片(因为缺少主分片)。
查看节点列表: http://ip:9200/_cat/nodes?v
curl -XPUT http://192.168.10.201:9200/user { "settings": { "number_of_shards": 3, "number_of_replicas": 1 }, "mappings": { "dynamic": "false", "properties": { "username": { "type": "keyword" }, "phone": { "type": "keyword", "index": "false" }, "description": { "type": "text" }, "create_time": { "type": "date", "format": "yyyy-MM-dd HH:mm:ss||yyyy-MM-dd||epoch_millis" } } } }
没有指定ID POST /index/_doc {}
POST user/_doc
{
"username": "zhangsan",
"phone": "123456",
"description": "勇敢、坚强、正直",
"create_time": "2022-08-05"
}
指定ID插入 POST /index/_doc/id {}
POST user/_doc/1
{
"username": "admin",
"phone": "123456",
"description": "管理员、至高无上、神一般",
"create_time": "2022-08-05"
}
查询索引 GET index/_search
GET user/_search
根据ID查询文档 GET index/_doc/id
GET user/_doc/1
查询指定的列 includes excludes
GET /user/_search
{
"_source": {
"includes": [
"username",
"description"
],
"excludes": [
"phone"
]
}
}
GET user/_doc/1
局部更新 POST index/_update/id
POST user/_update/1
{
"doc": {
"username": "test"
}
}
直接更新 PUT index/_doc/id
PUT user/_doc/1
{
"username": "test",
"phone": "123456"
}
# DELETE index/_doc/id
DELETE user/_doc/1
创建索引
PUT poetry { "settings": { "number_of_shards": 3, "number_of_replicas": 1 }, "mappings": { "dynamic": "false", "properties": { "name": { "type": "text", "analyzer": "ik_max_word", "search_analyzer": "ik_smart" }, "author": { "type": "keyword" }, "dynasty": { "type": "keyword" }, "content": { "type": "text", "analyzer": "ik_max_word", "search_analyzer": "ik_smart" } } } }
插入数据
POST poetry/_doc
{
"name": "望庐山瀑布",
"author": "李白",
"dynasty": "唐代",
"content": "日照香炉生紫烟,遥看瀑布挂前川。飞流直下三千尺,疑是银河落九天。"
}
组合条件查询
must、must not、should 区别:
==
and
!=
not
||
or
各类查询参数
term
{ "query": { "bool": { "must": [{ "term": { "content": "明月" } }], "must_not": [], "should": [] } }, "from": 0, "size": 10, "sort": [], "aggs": {} }
GET poetry/_search
{
"query": {
"term": {
"content": "明月"
}
}
}
match
{ "query": { "bool": { "must": [{ "match": { "content": "明月香炉" } }], "must_not": [], "should": [] } }, "from": 0, "size": 10, "sort": [], "aggs": {} }
GET poetry/_search
{
"query": {
"match": {
"content": "明月香炉"
}
}
}
prefix
GET poetry/_search
{
"query": {
"prefix": {
"content": "人生"
}
}
}
wildcard
GET poetry/_search
{
"query": {
"wildcard": {
"content": "生*"
}
}
}
fuzzy
fuzziness
的默认值是2 ——表示最多可以纠错两次fuzziness
的值太大, 将削弱检索条件的作用, 也就是说纠错次数太多, 就会导致限定检索结果的检索条件被改变, 失去了限定作用GET poetry/_search
{
"query": {
"fuzzy": {
"content": {
"value": "人生大事",
"fuzziness": 2
}
}
}
}
range
-1d/h/m/s
来进行时间操作query_string
int、long、string
查询,对 int、long 只能本身查询,对比 string 进行分词和本身查询missing
null
的文本一台服务器上无法存储大量数据,ES把一个index里面的数据分成多个shard分布式的存储在多个服务器上(对大的索引分片,拆成多个,分不到不同的节点上)。
ES就是通过shard来解决节点的容量上限问题的,通过主分片可以将数据分布到集群内的所有节点上。主分片数是在索引创建时指定的,一般不允许修改,除非Reindex。
一个索引中的数据保存在多个分片中(默认为一个)相当于水平分表。一个分片表示一个Lucene的实例,它本身就是一个完整的搜索引擎。我们的文档被存储和索引到分片内,这些对应用程序是透明的,即应用程序直接与索引交互而不是分片。
由上图可以看到shard可以分为主分片(primary shard)和副分片(replaca shard)。
图示为三个节点。其中任何一个都是有可能故障或者宕机的,此时其承载的shard的数据就会丢失;通过设置一个或者多个replica shard就可以在发生故障的时候提供备份服务。一方面可以保证数据不丢失,另一方面还可以提升操作的吞吐量和性能。
Reindex
over-sharding
的问题
Shard
和 Replication
的路由规则
Shard
组成,每个 Shard
有一个主节点和多个副本节点,副本个数可配置_routing
规则选择发给哪个 Shard
Index Request
中可以设置使用哪个 Filed
的值作为路由参数Mapping
中的配置_id
的 Hash
值选择出 Primary Shard
Replica Shard
数据安全策略
IO
保证读写性能,一般是每隔一段时间才会把 Lucene 的 Segment
写入磁盘持久化CommitLog
模块,Elasticsearch 中叫 TransLog
Shard
后,先写 Buffer
文件,创建好索引,此时索引还在内存里面,接着去写 TransLog
TransLog
后,刷新 TransLog
数据到磁盘上,写磁盘成功后,请求返回给用户CommitLog
,然后再写内存,而 Elasticsearch 是先将数据写入内存,最后才写 TransLog
Buffer
后,文档并不是可被搜索的,需要通过 Refresh
把内存的对象转化成完整的 Segment
后,然后再次 reopen
后才能被搜索
Segment
中的文档可以被搜索到,但是尚未被写入硬盘,即如果此时发送断电,则这些文档可能会丢失TransLog
写入磁盘,操作记录被写入磁盘 ES 才会将操作成功的结果返回发送给此操作请求的客户端TransLog
保证安全 =>(buffer
+ segment
)Segment
,Segment
文件越来越多(内存),而 TransLog
文件将越来越大(硬盘)TransLog
文件变得很大时,将执行一次 fsync
操作,此时所有在文件系统缓存中的 Segment
将被写入磁盘,而 TransLog
将被删除如果是每一次删除,都需要从集群中找到存储那个数据的节点进行删除,那么效率也太低了吧?
Elasticsearch 是这么操作的:
如果是删除请求的话,提交的时候会生成一个 .del
文件,里面将某个 doc
标识为 deleted
状态,那么搜索的时候根据 .del
文件就知道这个 doc
被删除了,客户端搜索的时候,发现数据在 .del
文件中标识为删除状态就不会被搜索到了
Update
请求后,从 Segment
或者 TransLog
中读取同 id
的完整 Doc
,记录版本号为 V1 = 345
V1
的全量 Doc
和请求中的部分字段 Doc
合并为一个完整的 Doc
,同时更新内存中的 VersionMap
Doc
后,Update
请求就变成了 POST/PUT
请求VersionMap
中读取该 id 的最大版本号 V2 = 346
V1 == V2
),如果冲突,则回退到开始的 Update doc
阶段重新执行,如果不冲突,则执行最新的 Add
请求Index Doc
阶段,首先将 version + 1
得到 V3
,再将 Doc
加入到 Lucene
中去,Lucene
中会先删除同 id
下的已存在 doc id
,然后再增加新的 Doc
。写入 Lucene
成功后,将当前 V3
更新到 VersionMap
中query
和取回 fetch
两个阶段from
名开始的数量为 size
的结果集,则每个节点都需要生成一个 from + size
大小的结果集TFIDF
BM25评分算法
near-real-time
(近实时)。Index 的实时性是由 refresh
控制的,默认是 1s,最快可到 100ms,那么也就意味着 index doc 成功后,需要等待一秒钟后才可以被搜索到。Get
请求也能保证是实时的,因为 Get 请求会直接读内存中尚未 Flush
到磁盘的 TransLog
。但是 Get
请求只支持通过 doc_id
进行查询,所以对于条件查询依然无法实现实时。NRT(Near Real Time)
近实时的系统NoSQL
数据库时,查询方式是 GetById
,这种查询可用直接从 TransLog
中查询,这时候就成了 RT(Real Time)
实时系统,只能根据 ID 进行查询rebuild
就可以了。TransLog
的 Flush
频率可以控制可靠性,要么是按请求,每次请求都 Flush
;要么是按时间,每隔一段时间 Flush
一次Flush
一次,Flush
间隔时间越长,可靠性就会越低Elasticsearch 有两种连接方式:transport
、rest
。
transport 通过 TCP
方式访问 ES(只支持java)
rest 方式通过 http API
访问 ES(没有语言限制)
ES 官方建议使用 rest 方式,transport
在 7.0
版本中不建议使用,在 8.x
的版本中废弃
你可以用 java 客户端做很多事:
执行标准的 index
、get
、delete
、update
、search
等操作
在正在运行的集群上执行管理任务
通过官方文档可以得知,现在存在至少三种 java 客户端:
造成这种混乱的原因是:
TransportClient
。但是 TransportClient
的缺点是显而易见的,它没有使用 RESRful
风格的接口,而是二进制的方式传输数据Java Low Level Rest Client
,它支持 RESTful
,用起来也不错,但是缺点也很明显,因为 TransportClient 的使用者把代码迁移到 Low Level Rest Client
的工作量比较大。官方文档专门为迁移代码出了一堆文档来提供参考Java High Level Rest Client
,它是基于 Java Low Level Rest Client
的封装,并且 API 接收参数和返回值与 TransportClient
是一样的,使得代码迁移变得容易并且支持了 RESTful
风格,兼容了这两种客户端的优点。当然缺点也是存在的,就是版本的问题,ES 的小版本更新非常频繁,在最理想的情况下,客户端的版本要和 ES 的版本一致(至少主版本号一致),次版本号不一致的话,基本操作也许可以,但是新的 API 就不支持了。下面介绍下 SpringBoot 如何通过 elasticsearch-rest-high-level-client 工具操作 ElasticSearch,这里需要说一下,为什么没有使用 Spring 家族封装的 spring-data-elasticsearch。
主要原因是灵活性和更新速度,Spring 将 ElasticSearch 过度封装,让开发者很难跟 ES 的 DSL 查询语句进行关联。再者就是更新速度,ES 的更新速度是非常快,但是 spring-data-elasticsearch 更新速度比较缓慢。
由于上面两点,所以选择了官方推出的 Java 客户端 elasticsearch-rest-high-level-client,它的代码写法跟 DSL 语句很相似,懂 ES 查询的使用其上手很快。
pom.xml
<!--json解析工具--> <dependency> <groupId>com.alibaba</groupId> <artifactId>fastjson</artifactId> <version>1.2.68</version> </dependency> <!-- 单元测试 --> <dependency> <groupId>junit</groupId> <artifactId>junit</artifactId> </dependency> <!-- Elasticsearch 服务端依赖 --> <dependency> <groupId>org.elasticsearch</groupId> <artifactId>elasticsearch</artifactId> <version>7.17.2</version> </dependency> <!-- rest-client 客户端依赖 --> <dependency> <groupId>org.elasticsearch.client</groupId> <artifactId>elasticsearch-rest-client</artifactId> <version>7.17.2</version> </dependency> <!-- rest-high-level-client 客户端依赖 --> <dependency> <groupId>org.elasticsearch.client</groupId> <artifactId>elasticsearch-rest-high-level-client</artifactId> <version>7.17.2</version> </dependency>
application.yml
配置文件
elasticsearch:
schema: http
address: 192.168.10.201:9200,192.168.10.202:9200,192.168.10.203:9200
connectTimeout: 5000
soketTimeout: 5000
connectionRequestTimeout: 5000
maxConnectNum: 100
maxConnectPerRoute: 100
java 连接配置类:
package com.moon.es.core.configuration; import org.apache.http.HttpHost; import org.elasticsearch.client.RestClient; import org.elasticsearch.client.RestClientBuilder; import org.elasticsearch.client.RestHighLevelClient; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import java.util.ArrayList; import java.util.List; @Configuration public class ElasticsearchConfiguration { /** 协议 */ @Value("${elasticsearch.schema:http}") private String schema; /** 集群地址,如果有多个用 "," 隔开 */ @Value("${elasticsearch.address}") private String address; /** 连接超时时间 */ @Value("${elasticsearch.connectTimeout}") private Integer connectTimeout; /** socket 连接超时时间 */ @Value("${elasticsearch.socketTimeout}") private Integer socketTimeout; /** 获取连接的超时时间 */ @Value("${elasticsearch.connectionRequestTimeout}") private Integer connectionRequestTimeout; /** 最大连接数 */ @Value("${elasticsearch.maxConnectNum}") private Integer maxConnectNum; /** 最大路由连接数 */ @Value("${elasticsearch.maxConnectPerRoute}") private Integer maxConnectPerRoute; @Bean public RestHighLevelClient restHighLevelClient() { // 拆分地址 List<HttpHost> hostList = new ArrayList<>(); String[] hosts = this.address.split(","); for (String address : hosts) { String host = address.split(":")[0]; String port = address.split(":")[1]; hostList.add(new HttpHost(host, Integer.parseInt(port), this.schema)); } // 转换成 HttpHost 数组 HttpHost[] httpHosts = hostList.toArray(new HttpHost[]{}); // 构建连接对象 RestClientBuilder builder = RestClient.builder(httpHosts); // 异步连接延时配置 builder.setRequestConfigCallback(requestConfigBuilder -> { requestConfigBuilder.setConnectTimeout(this.connectTimeout); requestConfigBuilder.setSocketTimeout(this.socketTimeout); requestConfigBuilder.setConnectionRequestTimeout(this.connectionRequestTimeout); return requestConfigBuilder; }); // 异步连接数配置 builder.setHttpClientConfigCallback(httpClientBuilder -> { httpClientBuilder.setMaxConnTotal(this.maxConnectNum); httpClientBuilder.setMaxConnPerRoute(this.maxConnectPerRoute); return httpClientBuilder; }); return new RestHighLevelClient(builder); } }
创建名为 user 的索引与对应 Mapping
PUT /user { "settings": { "number_of_shards": 3, "number_of_replicas": 1 }, "mappings": { "dynamic": true, "properties": { "name": { "type": "text", "analyzer": "ik_max_word", "search_analyzer": "ik_smart", "fields": { "keyword": { "type": "keyword" } } }, "address": { "type": "text", "analyzer": "ik_max_word", "search_analyzer": "ik_smart", "fields": { "keyword": { "type": "keyword" } } }, "remark": { "type": "text", "analyzer": "ik_max_word", "search_analyzer": "ik_smart" }, "age": { "type": "integer" }, "salary": { "type": "float" }, "birthDate": { "type": "date", "format": "yyyy-MM-dd" }, "createTime": { "type": "date" } } } }
代码示例:
package com.moon.es.service.impl; import lombok.extern.slf4j.Slf4j; import com.moon.es.service.IndexService; import org.elasticsearch.action.admin.indices.create.CreateIndexRequest; import org.elasticsearch.action.admin.indices.create.CreateIndexResponse; import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest; import org.elasticsearch.action.support.master.AcknowledgedResponse; import org.elasticsearch.client.RequestOptions; import org.elasticsearch.client.RestHighLevelClient; import org.elasticsearch.common.settings.Settings; import org.elasticsearch.xcontent.XContentBuilder; import org.elasticsearch.xcontent.XContentFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; import java.io.IOException; @Slf4j @Service public class IndexServiceImpl implements IndexService { @Autowired private RestHighLevelClient restHighLevelClient; @Value("${elasticsearch.indexName}") private String indexName; /** * 创建索引 * @return boolean */ @Override public boolean createIndex() { boolean result = false; try { // 创建 Mapping XContentBuilder mapping = XContentFactory.jsonBuilder() .startObject() .field("dynamic", true) .startObject("properties") .startObject("name") .field("type","text") .field("analyzer", "ik_max_word") .field("search_analyzer", "ik_smart") .startObject("fields") .startObject("keyword") .field("type","keyword") .endObject() .endObject() .endObject() .startObject("address") .field("type","text") .field("analyzer", "ik_max_word") .field("search_analyzer", "ik_smart") .startObject("fields") .startObject("keyword") .field("type","keyword") .endObject() .endObject() .endObject() .startObject("remark") .field("type","text") .field("analyzer", "ik_max_word") .field("search_analyzer", "ik_smart") .endObject() .startObject("age") .field("type","integer") .endObject() .startObject("salary") .field("type","float") .endObject() .startObject("birthDate") .field("type","date") .field("format", "yyyy-MM-dd") .endObject() .startObject("createTime") .field("type","date") .endObject() .endObject() .endObject(); // 创建索引配置信息 Settings settings = Settings.builder() .put("index.number_of_shards", 3) .put("index.number_of_replicas", 1) .build(); // 新建创建索引请求对象,然后设置索引类型(ES 7.0 将不存在索引类型)和 mapping 与 index 配置 CreateIndexRequest request = new CreateIndexRequest(this.indexName, settings); request.mapping("_doc", mapping); // RestHighLevelClient 执行创建索引 CreateIndexResponse createIndexResponse = restHighLevelClient.indices().create(request, RequestOptions.DEFAULT); result = createIndexResponse.isAcknowledged(); log.info("是否创建 index 成功: {}", result); } catch (IOException e) { log.error("创建 index 异常", e); } return result; } /** * 删除索引 * @return boolean */ @Override public boolean deleteIndex() { boolean result = false; try { // 新建删除索引请求对象 DeleteIndexRequest request = new DeleteIndexRequest(this.indexName); // 执行删除索引 AcknowledgedResponse response = restHighLevelClient.indices().delete(request, RequestOptions.DEFAULT); result = response.isAcknowledged(); log.info("是否删除 index 成功: {}", result); } catch (IOException e) { log.error("删除 index 异常", e); } return result; } }
单元测试:
package com.moon.es; import com.moon.es.service.IndexService; import org.junit.Assert; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import org.springframework.test.context.web.WebAppConfiguration; @WebAppConfiguration @RunWith(SpringJUnit4ClassRunner.class) @SpringBootTest public class ElasticsearchTest { @Autowired private IndexService indexService; @Test public void indexCreate() { boolean result = this.indexService.createIndex(); Assert.assertTrue("创建索引成功!", result); } @Test public void indexDelete() { boolean result = this.indexService.deleteIndex(); Assert.assertTrue("删除索引成功!", result); } }
// 新增文档信息 POST /user/ { "address": "北京市", "age": 29, "birthDate": "1990-01-10", "createTime": 1579530727699, "name": "张三", "remark": "来自北京市的张先生", "salary": 100 } // 获取文档信息 GET /user/doc/1 // 更新文档信息 PUT /user/1 { "address": "北京市海淀区", "age": 29, "birthDate": "1990-01-10", "createTime": 1579530727699, "name": "张三", "remark": "来自北京市的张先生", "salary": 100 } // 删除文档信息 DELETE /user/1
实体类:
package com.moon.es.entity; import lombok.*; import java.math.BigDecimal; import java.util.Date; @Getter @Setter @Builder @ToString @NoArgsConstructor @AllArgsConstructor public class User { private String name; private Integer age; private BigDecimal salary; private String address; private String remark; private String birthDate; private Date createTime; }
代码示例:
package com.moon.es.service.impl; import com.alibaba.fastjson.JSON; import com.moon.es.entity.User; import com.moon.es.service.DocumentService; import lombok.extern.slf4j.Slf4j; import org.elasticsearch.action.delete.DeleteRequest; import org.elasticsearch.action.delete.DeleteResponse; import org.elasticsearch.action.get.GetRequest; import org.elasticsearch.action.get.GetResponse; import org.elasticsearch.action.index.IndexRequest; import org.elasticsearch.action.index.IndexResponse; import org.elasticsearch.action.update.UpdateRequest; import org.elasticsearch.action.update.UpdateResponse; import org.elasticsearch.client.RequestOptions; import org.elasticsearch.client.RestHighLevelClient; import org.elasticsearch.xcontent.XContentType; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; import java.io.IOException; import java.math.BigDecimal; import java.util.Date; @Service @Slf4j public class DocumentServiceImpl implements DocumentService { @Autowired private RestHighLevelClient restHighLevelClient; @Value("${elasticsearch.indexName}") private String indexName; /** * 新增文档 * @return status */ @Override public int addDocument() { int status = 400; try { // 创建索引请求对象 IndexRequest request = new IndexRequest(this.indexName, "_doc"); User user = User.builder() .name("张三") .age(29) .salary(new BigDecimal("100.00")) .address("北京市") .remark("来自北京市的张先生") .birthDate("1990-01-10") .createTime(new Date()) .build(); // 将对象转换为 byte 数组 byte[] json = JSON.toJSONBytes(user); // 设置文档内容 request.source(json, XContentType.JSON); IndexResponse response = restHighLevelClient.index(request, RequestOptions.DEFAULT); status = response.status().getStatus(); log.info("新增文档状态: {}", status); } catch (IOException e) { log.error("新增文档异常", e); } return status; } /** * 获取文档信息 * @param id 文档id * @return user */ @Override public User getDocument(String id) { User user = null; try { GetRequest request = new GetRequest(this.indexName, "_doc", id); GetResponse response = restHighLevelClient.get(request, RequestOptions.DEFAULT); if (response.isExists()) { user = JSON.parseObject(response.getSourceAsBytes(), User.class); log.info("获取文档信息: {}", user); } } catch (IOException e) { log.error("获取文档信息异常", e); } return user; } /** * 更新文档信息 * @param id 文档id * @return status */ @Override public int updateDocument(String id) { int status = 400; try { UpdateRequest request = new UpdateRequest(this.indexName, "_doc", id); User user = User.builder() .salary(new BigDecimal("200.00")) .address("北京市海淀区") .build(); byte[] json = JSON.toJSONBytes(user); request.doc(json, XContentType.JSON); UpdateResponse response = restHighLevelClient.update(request, RequestOptions.DEFAULT); status = response.status().getStatus(); log.info("更新文档状态: {}", status); } catch (IOException e) { log.error("更新文档信息异常", e); } return status; } /** * 删除文档信息 * @param id 文档id * @return status */ @Override public int deleteDocument(String id) { int status = 400; try { DeleteRequest request = new DeleteRequest(this.indexName, "_doc", id); DeleteResponse response = restHighLevelClient.delete(request, RequestOptions.DEFAULT); status = response.status().getStatus(); log.info("删除文档状态: {}", status); } catch (IOException e) { log.error("删除文档信息异常", e); } return status; } }
单元测试:
@Test public void docCreate() { int status = this.documentService.addDocument(); Assert.assertEquals("新增文档成功!", 201, status); } @Test public void docQuery() { User user = this.documentService.getDocument("1"); Assert.assertNotNull("查询文档成功!", user); } @Test public void docUpdate() { int status = this.documentService.updateDocument("1"); Assert.assertEquals("更新文档成功!", 200, status); } @Test public void docDelete() { int status = this.documentService.deleteDocument("1"); Assert.assertEquals("删除文档成功!", 200, status); }
// 精确查询 // 查询条件不会进行分词,但是查询内容可能会分词,导致查询不到 // 之前在创建索引时设置 Mapping 中 address 字段存在 keyword 字段是专门用于不分词查询的子字段。 GET /user/_search { "query": { "term": { "address.keyword": { "value": "北京市通州区" } } } } // 多内容查询精确查询 GET /user/_search { "query": { "terms": { "address.keyword": [ "北京市丰台区", "北京市昌平区", "北京市大兴区" ] } } }
代码示例:
package com.moon.es.service.impl; import com.alibaba.fastjson.JSON; import com.moon.es.entity.User; import com.moon.es.service.TermQueryService; import lombok.extern.slf4j.Slf4j; import org.elasticsearch.action.search.SearchRequest; import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.client.RequestOptions; import org.elasticsearch.client.RestHighLevelClient; import org.elasticsearch.index.query.QueryBuilders; import org.elasticsearch.rest.RestStatus; import org.elasticsearch.search.SearchHit; import org.elasticsearch.search.SearchHits; import org.elasticsearch.search.builder.SearchSourceBuilder; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; import java.io.IOException; import java.util.ArrayList; import java.util.List; @Slf4j @Service public class TermQueryServiceImpl implements TermQueryService { @Autowired private RestHighLevelClient restHighLevelClient; @Value("${elasticsearch.indexName}") private String indexName; /** * 精确查询 * 查询条件不会进行分词,但是查询内容可能会分词,导致查询不到 * @param field 查询字段 * @param keyword 查询关键词 * @return userList */ @Override public List<User> termQuery(String field, String keyword) { List<User> userList = new ArrayList<>(); SearchSourceBuilder sourceBuilder = new SearchSourceBuilder(); sourceBuilder.query(QueryBuilders.termQuery(field, keyword)); SearchRequest request = new SearchRequest(this.indexName); request.source(sourceBuilder); try { SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT); // 根据状态和数据条数验证是否返回了数据 if (response.status().equals(RestStatus.OK) && response.getHits().getTotalHits().value > 0) { SearchHits hits = response.getHits(); for (SearchHit hit : hits) { User user = JSON.parseObject(hit.getSourceAsString(), User.class); userList.add(user); } log.info("精确查询成功: {}", userList); } } catch (IOException e) { log.error("精确查询异常", e); } return userList; } /** * 多个内容在一个字段中进行查询 * @param field 查询字段 * @param keywords 多个查询关键字 * @return userList */ @Override public List<User> termsQuery(String field, String... keywords) { List<User> userList = new ArrayList<>(); SearchSourceBuilder sourceBuilder = new SearchSourceBuilder(); sourceBuilder.query(QueryBuilders.termsQuery(field, keywords)); SearchRequest request = new SearchRequest(this.indexName); request.source(sourceBuilder); try { SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT); if (response.status().equals(RestStatus.OK) && response.getHits().getTotalHits().value > 0) { SearchHits hits = response.getHits(); for (SearchHit hit : hits) { User user = JSON.parseObject(hit.getSourceAsString(), User.class); userList.add(user); } log.info("多内容精确查询成功: {}", userList); } } catch (IOException e) { log.error("多内容精确查询异常", e); } return userList; } }
单元测试:
@Test
public void termQuery() {
List<User> userList = this.termQueryService.termQuery("address.keyword", "北京市通州区");
Assert.assertNotEquals("精确查询成功!", 0, userList.size());
}
@Test
public void termsQuery() {
List<User> userList = this.termQueryService.termsQuery("address.keyword", "北京市海淀区", "北京市昌平区", "北京市大兴区");
Assert.assertNotEquals("多内容精确查询成功!", 0, userList.size());
}
// 匹配查询全部数据与分页 // 匹配查询符合条件的所有数据,并且设置以 salary 字段升序排序,并设置分页 GET /user/_search { "query": { "match_all": {} }, "from": 0, "size": 10, "sort": [ { "salary": { "order": "asc" } } ] } // 匹配查询数据 GET /user/_search { "query": { "match": { "address": "*通州区" } } } //词语匹配查询 GET user/_search { "query": { "match_phrase": { "address": "通州区" } } } // 内容多字段查询 GET user/_search { "query": { "multi_match": { "query": "北京", "fields": [ "address", "remark" ] } } }
示例代码:
package com.moon.es.service.impl; import com.alibaba.fastjson.JSON; import com.moon.es.entity.User; import com.moon.es.service.MatchQueryService; import lombok.extern.slf4j.Slf4j; import org.elasticsearch.action.search.SearchRequest; import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.client.RequestOptions; import org.elasticsearch.client.RestHighLevelClient; import org.elasticsearch.index.query.QueryBuilders; import org.elasticsearch.rest.RestStatus; import org.elasticsearch.search.SearchHit; import org.elasticsearch.search.SearchHits; import org.elasticsearch.search.builder.SearchSourceBuilder; import org.elasticsearch.search.sort.SortOrder; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; import java.io.IOException; import java.util.ArrayList; import java.util.List; @Slf4j @Service public class MatchQueryServiceImpl implements MatchQueryService { @Autowired private RestHighLevelClient restHighLevelClient; @Value("${elasticsearch.indexName}") private String indexName; /** * 匹配查询符合条件的所有数据,并设置分页 * @param from 分页偏移量 * @param size 每页条数 * @param sortField 排序字段 * @return userList */ @Override public List<User> matchAllQuery(Integer from, Integer size, String sortField) { List<User> userList = new ArrayList<>(); SearchSourceBuilder sourceBuilder = new SearchSourceBuilder(); sourceBuilder.query(QueryBuilders.matchAllQuery()); sourceBuilder.from(from); sourceBuilder.size(size); sourceBuilder.sort(sortField, SortOrder.ASC); SearchRequest request = new SearchRequest(this.indexName); request.source(sourceBuilder); try { SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT); if (response.status().equals(RestStatus.OK) && response.getHits().getTotalHits().value > 0) { SearchHits hits = response.getHits(); for (SearchHit hit : hits) { User user = JSON.parseObject(hit.getSourceAsString(), User.class); userList.add(user); } log.info("全匹配查询: {}", userList); } } catch (IOException e) { log.error("全匹配查询异常", e); } return userList; } /** * 匹配查询数据 * @param field 查询字段 * @param keyword 查询关键字 * @return userList */ @Override public List<User> matchQuery(String field, String keyword) { List<User> userList = new ArrayList<>(); SearchSourceBuilder sourceBuilder = new SearchSourceBuilder(); sourceBuilder.query(QueryBuilders.matchQuery(field, keyword)); SearchRequest request = new SearchRequest(this.indexName); request.source(sourceBuilder); try { SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT); if (response.status().equals(RestStatus.OK) && response.getHits().getTotalHits().value > 0) { SearchHits hits = response.getHits(); for (SearchHit hit : hits) { User user = JSON.parseObject(hit.getSourceAsString(), User.class); userList.add(user); } log.info("匹配查询: {}", userList); } } catch (IOException e) { log.error("匹配查询异常", e); } return userList; } /** * 词语匹配查询 * @param field 查询字段 * @param keyword 查询关键词 * @return userList */ @Override public List<User> matchPhraseQuery(String field, String keyword) { List<User> userList = new ArrayList<>(); SearchSourceBuilder sourceBuilder = new SearchSourceBuilder(); sourceBuilder.query(QueryBuilders.matchPhraseQuery(field, keyword)); SearchRequest request = new SearchRequest(this.indexName); request.source(sourceBuilder); try { SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT); if (response.status().equals(RestStatus.OK) && response.getHits().getTotalHits().value > 0) { SearchHits hits = response.getHits(); for (SearchHit hit : hits) { User user = JSON.parseObject(hit.getSourceAsString(), User.class); userList.add(user); } log.info("词语匹配查询: {}", userList); } } catch (IOException e) { log.error("词语匹配查询异常", e); } return userList; } /** * 内容在多字段中进行查询 * @param keyword 查询关键字 * @param fields 查询多字段 * @return userList */ @Override public List<User> matchMultiQuery(String keyword, String... fields) { List<User> userList = new ArrayList<>(); SearchSourceBuilder sourceBuilder = new SearchSourceBuilder(); sourceBuilder.query(QueryBuilders.multiMatchQuery(keyword, fields)); SearchRequest request = new SearchRequest(this.indexName); request.source(sourceBuilder); try { SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT); if (response.status().equals(RestStatus.OK) && response.getHits().getTotalHits().value > 0) { SearchHits hits = response.getHits(); for (SearchHit hit : hits) { User user = JSON.parseObject(hit.getSourceAsString(), User.class); userList.add(user); } log.info("多字段查询: {}", userList); } } catch (IOException e) { log.error("多字段查询异常", e); } return userList; } }
单元测试:
@Test public void matchAllQuery() { List<User> userList = this.matchQueryService.matchAllQuery(0, 10, "salary"); Assert.assertNotEquals("全匹配查询成功!", 0, userList.size()); } @Test public void matchQuery() { List<User> userList = this.matchQueryService.matchQuery("address", "*通州区"); Assert.assertNotEquals("匹配查询成功!", 0, userList.size()); } @Test public void matchPhraseQuery() { List<User> userList = this.matchQueryService.matchPhraseQuery("address", "通州区"); Assert.assertNotEquals("词语匹配查询成功!", 0, userList.size()); } @Test public void matchMultiQuery() { List<User> userList = this.matchQueryService.matchMultiQuery("李先生大兴区", "address", "remark"); Assert.assertNotEquals("词语匹配查询成功!", 0, userList.size()); }
// 模糊查询所有以 "三" 结尾的姓名
GET /user/_search
{
"query": {
"fuzzy": {
"name": "三"
}
}
}
代码示例:
package com.moon.es.service.impl; import com.alibaba.fastjson.JSON; import com.moon.es.entity.User; import com.moon.es.service.FuzzyQueryService; import lombok.extern.slf4j.Slf4j; import org.elasticsearch.action.search.SearchRequest; import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.client.RequestOptions; import org.elasticsearch.client.RestHighLevelClient; import org.elasticsearch.common.unit.Fuzziness; import org.elasticsearch.index.query.QueryBuilders; import org.elasticsearch.rest.RestStatus; import org.elasticsearch.search.SearchHit; import org.elasticsearch.search.SearchHits; import org.elasticsearch.search.builder.SearchSourceBuilder; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; import java.io.IOException; import java.util.ArrayList; import java.util.List; @Slf4j @Service public class FuzzyQueryServiceImpl implements FuzzyQueryService { @Autowired private RestHighLevelClient restHighLevelClient; @Value("${elasticsearch.indexName}") private String indexName; /** * 模糊查询所有以 [keyword] 结尾的文档 * @param field 查询字段 * @param keyword 查询关键字 * @return userList */ @Override public List<User> fuzzyQuery(String field, String keyword) { List<User> userList = new ArrayList<>(); SearchSourceBuilder sourceBuilder = new SearchSourceBuilder(); sourceBuilder.query(QueryBuilders.fuzzyQuery(field, keyword).fuzziness(Fuzziness.AUTO)); SearchRequest request = new SearchRequest(this.indexName); request.source(sourceBuilder); try { SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT); if (response.status().equals(RestStatus.OK) && response.getHits().getTotalHits().value > 0) { SearchHits hits = response.getHits(); for (SearchHit hit : hits) { User user = JSON.parseObject(hit.getSourceAsString(), User.class); userList.add(user); } log.info("模糊查询: {}", userList); } } catch (IOException e) { log.error("模糊查询异常", e); } return userList; } }
单元测试:
@Test
public void fuzzyQuery() {
List<User> userList = this.fuzzyQueryService.fuzzyQuery("name", "三");
Assert.assertNotEquals("模糊查询成功!", 0, userList.size());
}
// 查询岁数 ≥ 30 岁的员工数据: GET /user/_search { "query": { "range": { "age": { "gte": 30 } } } } // 查询生日距离现在 30 年间的员工数据: GET /user/_search { "query": { "range": { "birthDate": { "gte": "now-30y" } } } }
代码示例:
package com.moon.es.service.impl; import com.alibaba.fastjson.JSON; import com.moon.es.entity.User; import com.moon.es.service.RangeQueryService; import lombok.extern.slf4j.Slf4j; import org.elasticsearch.action.search.SearchRequest; import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.client.RequestOptions; import org.elasticsearch.client.RestHighLevelClient; import org.elasticsearch.index.query.QueryBuilders; import org.elasticsearch.rest.RestStatus; import org.elasticsearch.search.SearchHit; import org.elasticsearch.search.SearchHits; import org.elasticsearch.search.builder.SearchSourceBuilder; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; import java.io.IOException; import java.util.ArrayList; import java.util.List; @Slf4j @Service public class RangeQueryServiceImpl implements RangeQueryService { @Autowired private RestHighLevelClient restHighLevelClient; @Value("${elasticsearch.indexName}") private String indexName; /** * 查询岁数 ≥ [value] 岁的员工数据 * @param field 查询字段 * @param value 范围值 * @return userList */ @Override public List<User> rangeGteQuery(String field, Integer value) { List<User> userList = new ArrayList<>(); SearchSourceBuilder sourceBuilder = new SearchSourceBuilder(); sourceBuilder.query(QueryBuilders.rangeQuery(field).gte(value)); SearchRequest request = new SearchRequest(this.indexName); request.source(sourceBuilder); try { SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT); if (response.status().equals(RestStatus.OK) && response.getHits().getTotalHits().value > 0) { SearchHits hits = response.getHits(); for (SearchHit hit : hits) { User user = JSON.parseObject(hit.getSourceAsString(), User.class); userList.add(user); } log.info("gte 范围查询: {}", userList); } } catch (IOException e) { log.error("gte 范围查询异常", e); } return userList; } /** * 查询距离现在 [value] 年间的员工数据 * [年(y)、月(M)、星期(w)、天(d)、小时(h)、分钟(m)、秒(s)] * now-1h 查询一小时内范围 * now-1d 查询一天内时间范围 * now-1y 查询最近一年内的时间范围 * @param field 查询字段 * @param value 范围值 * @return userList */ @Override public List<User> rangeDateQuery(String field, String value) { List<User> userList = new ArrayList<>(); SearchSourceBuilder sourceBuilder = new SearchSourceBuilder(); // includeLower(是否包含下边界)、includeUpper(是否包含上边界) sourceBuilder.query(QueryBuilders.rangeQuery(field) .gte(value).includeLower(true).includeUpper(true)); SearchRequest request = new SearchRequest(this.indexName); request.source(sourceBuilder); try { SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT); if (response.status().equals(RestStatus.OK) && response.getHits().getTotalHits().value > 0) { SearchHits hits = response.getHits(); for (SearchHit hit : hits) { User user = JSON.parseObject(hit.getSourceAsString(), User.class); userList.add(user); } log.info("date 范围查询: {}", userList); } } catch (IOException e) { log.error("date 范围查询异常", e); } return userList; } }
单元测试:
@Test
public void rangeGteQuery() {
List<User> userList = rangeQueryService.rangeGteQuery("age", 30);
Assert.assertNotEquals("gte 范围查询成功!", 0, userList.size());
}
@Test
public void rangeDateQuery() {
List<User> userList = rangeQueryService.rangeDateQuery("birthDate", "now-30y");
Assert.assertNotEquals("date 范围查询成功!", 0, userList.size());
}
// 查询所有以 "张" 开头的姓名:
GET /user/_search
{
"query": {
"wildcard": {
"name": {
"value": "张*"
}
}
}
}
代码示例:
package com.moon.es.service.impl; import com.alibaba.fastjson.JSON; import com.moon.es.entity.User; import com.moon.es.service.WildcardQueryService; import lombok.extern.slf4j.Slf4j; import org.elasticsearch.action.search.SearchRequest; import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.client.RequestOptions; import org.elasticsearch.client.RestHighLevelClient; import org.elasticsearch.index.query.QueryBuilders; import org.elasticsearch.rest.RestStatus; import org.elasticsearch.search.SearchHit; import org.elasticsearch.search.SearchHits; import org.elasticsearch.search.builder.SearchSourceBuilder; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; import java.io.IOException; import java.util.ArrayList; import java.util.List; @Slf4j @Service public class WildcardQueryServiceImpl implements WildcardQueryService { @Autowired private RestHighLevelClient restHighLevelClient; @Value("${elasticsearch.indexName}") private String indexName; /** * 查询以 [keyword] 开头/结尾的文档 * *: 表示多个字符(0个或多个字符) * ?: 表示单个字符 * @param field 查询字段 * @param keyword 查询关键字 * @return userList */ @Override public List<User> wildcardQuery(String field, String keyword) { List<User> userList = new ArrayList<>(); SearchSourceBuilder sourceBuilder = new SearchSourceBuilder(); sourceBuilder.query(QueryBuilders.wildcardQuery(field, keyword)); SearchRequest request = new SearchRequest(this.indexName); request.source(sourceBuilder); try { SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT); if (response.status().equals(RestStatus.OK) && response.getHits().getTotalHits().value > 0) { SearchHits hits = response.getHits(); for (SearchHit hit : hits) { User user = JSON.parseObject(hit.getSourceAsString(), User.class); userList.add(user); } log.info("通配符查询: {}", userList); } } catch (IOException e) { log.error("通配符查询异常", e); } return userList; } }
单元测试:
@Test
public void wildcardQuery() {
List<User> userList = this.wildcardQueryService.wildcardQuery("name", "张*");
Assert.assertNotEquals("通配符查询成功!", 0, userList.size());
}
// 查询出生在 1990-1995 年期间,且地址在 北京市昌平区、北京市大兴区、北京市房山区 的员工信息: GET /user/_search { "query": { "bool": { "filter": { "range": { "birthDate": { "format": "yyyy", "gte": 1990, "lte": 1995 } } }, "must": [ { "terms": { "address.keyword": [ "北京市昌平区", "北京市大兴区", "北京市房山区" ] } } ] } } }
代码示例:
package com.moon.es.service.impl; import com.alibaba.fastjson.JSON; import com.moon.es.entity.User; import com.moon.es.service.BoolQueryService; import lombok.extern.slf4j.Slf4j; import org.elasticsearch.action.search.SearchRequest; import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.client.RequestOptions; import org.elasticsearch.client.RestHighLevelClient; import org.elasticsearch.index.query.BoolQueryBuilder; import org.elasticsearch.index.query.QueryBuilders; import org.elasticsearch.rest.RestStatus; import org.elasticsearch.search.SearchHit; import org.elasticsearch.search.SearchHits; import org.elasticsearch.search.builder.SearchSourceBuilder; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; import java.io.IOException; import java.util.ArrayList; import java.util.List; @Slf4j @Service public class BoolQueryServiceImpl implements BoolQueryService { @Autowired private RestHighLevelClient restHighLevelClient; @Value("${elasticsearch.indexName}") private String indexName; /** * 查询出生在 1990-1995 年期间,且地址在 北京市昌平区、北京市大兴区、北京市房山区 的员工信息 * @return userList */ @Override public List<User> boolQuery() { List<User> userList = new ArrayList<>(); // 创建 Bool 查询构建器 BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery(); queryBuilder.must(QueryBuilders.termsQuery("address.keyword", "北京市昌平区", "北京市大兴区", "北京市房山区")) .filter().add(QueryBuilders.rangeQuery("birthDate").format("yyyy").gte("1990").lte("1995")); // 构建查询源构建器 SearchSourceBuilder sourceBuilder = new SearchSourceBuilder(); sourceBuilder.query(queryBuilder); // 创建查询请求对象,将查询对象配置到其中 SearchRequest request = new SearchRequest(this.indexName); request.source(sourceBuilder); // 执行查询,然后处理响应结果 try { SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT); if (response.status().equals(RestStatus.OK) && response.getHits().getTotalHits().value > 0) { SearchHits hits = response.getHits(); for (SearchHit hit : hits) { User user = JSON.parseObject(hit.getSourceAsString(), User.class); userList.add(user); } log.info("布尔查询: {}", userList); } } catch (IOException e) { log.error("布尔查询异常", e); } return userList; } }
单元测试:
@Test
public void boolQuery() {
List<User> userList = this.boolQueryService.boolQuery();
Assert.assertNotEquals("布尔查询成功!", 0, userList.size());
}
// 统计员工总数、工资最高值、工资最低值、工资平均工资、工资总和: GET /user/_search { "size": 0, "aggs": { "salary_stats": { "stats": { "field": "salary" } } } } // 统计员工工资最低值: GET /user/_search { "size": 0, "aggs": { "salary_min": { "min": { "field": "salary" } } } } // 统计员工工资最高值: GET /user/_search { "size": 0, "aggs": { "salary_max": { "max": { "field": "salary" } } } } // 统计员工工资平均值: GET /user/_search { "size": 0, "aggs": { "salary_sum": { "sum": { "field": "salary" } } } } // 统计员工工资总值: GET /user/_search { "size": 0, "aggs": { "salary_sum": { "sum": { "field": "salary" } } } } // 统计员工总数: GET /user/_search { "size": 0, "aggs": { "employee_count": { "value_count": { "field": "salary" } } } } // 统计员工工资百分位: GET /user/_search { "size": 0, "aggs": { "salary_percentiles": { "percentiles": { "field": "salary" } } } }
示例代码:
package com.moon.es.service.impl; import com.moon.es.entity.Stats; import com.moon.es.service.AggsMetricService; import lombok.extern.slf4j.Slf4j; import org.elasticsearch.action.search.SearchRequest; import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.client.RequestOptions; import org.elasticsearch.client.RestHighLevelClient; import org.elasticsearch.rest.RestStatus; import org.elasticsearch.search.aggregations.AggregationBuilder; import org.elasticsearch.search.aggregations.AggregationBuilders; import org.elasticsearch.search.aggregations.Aggregations; import org.elasticsearch.search.aggregations.metrics.*; import org.elasticsearch.search.builder.SearchSourceBuilder; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; import java.io.IOException; import java.math.BigDecimal; import java.util.HashMap; import java.util.Map; @Slf4j @Service public class AggsMetricServiceImpl implements AggsMetricService { @Autowired private RestHighLevelClient restHighLevelClient; @Value("${elasticsearch.indexName}") private String indexName; /** * stats 统计员工总数、员工工资最高值、员工工资最低值、员工平均工资、员工工资总和 * * @return {@link Stats} */ @Override public Stats aggregationStats() { Stats stats = null; // 设置聚合条件 AggregationBuilder aggregationBuilder = AggregationBuilders.stats("salary_stats").field("salary"); SearchSourceBuilder sourceBuilder = new SearchSourceBuilder(); sourceBuilder.aggregation(aggregationBuilder); // 设置查询结果不返回,只返回聚合结果 sourceBuilder.size(0); SearchRequest request = new SearchRequest(this.indexName); request.source(sourceBuilder); try { SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT); // 获取响应中的聚合信息 Aggregations aggs = response.getAggregations(); if (response.status().equals(RestStatus.OK) && aggs != null) { // 转换为 Stats 对象 ParsedStats parsedStats = aggs.get("salary_stats"); stats = Stats.builder() .count(parsedStats.getCount()) .min(BigDecimal.valueOf(parsedStats.getMin())) .max(BigDecimal.valueOf(parsedStats.getMax())) .avg(BigDecimal.valueOf(parsedStats.getAvg())) .sum(BigDecimal.valueOf(parsedStats.getSum())) .build(); } log.info("stats 度量聚合查询: {}", stats); } catch (IOException e) { log.error("stats 度量聚合查询异常", e); } return stats; } /** * min 统计员工工资最低值 * * @return {@link BigDecimal} */ @Override public BigDecimal aggregationMin() { BigDecimal min = BigDecimal.valueOf(-1); AggregationBuilder aggregationBuilder = AggregationBuilders.min("salary_min").field("salary"); SearchSourceBuilder sourceBuilder = new SearchSourceBuilder(); sourceBuilder.aggregation(aggregationBuilder); sourceBuilder.size(0); SearchRequest request = new SearchRequest(this.indexName); request.source(sourceBuilder); try { SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT); Aggregations aggs = response.getAggregations(); if (response.status().equals(RestStatus.OK) && aggs != null) { ParsedMin parsedMin = aggs.get("salary_min"); min = BigDecimal.valueOf(parsedMin.getValue()); } log.info("min 度量聚合查询: {}", min); } catch (IOException e) { log.error("min 度量聚合查询异常", e); } return min; } /** * max 统计员工工资最高值 * * @return {@link BigDecimal} */ @Override public BigDecimal aggregationMax() { BigDecimal max = BigDecimal.valueOf(-1); AggregationBuilder aggregationBuilder = AggregationBuilders.max("salary_max").field("salary"); SearchSourceBuilder sourceBuilder = new SearchSourceBuilder(); sourceBuilder.aggregation(aggregationBuilder); sourceBuilder.size(0); SearchRequest request = new SearchRequest(this.indexName); request.source(sourceBuilder); try { SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT); Aggregations aggs = response.getAggregations(); if (response.status().equals(RestStatus.OK) && aggs != null) { ParsedMax parsedMax = aggs.get("salary_max"); max = BigDecimal.valueOf(parsedMax.getValue()); } log.info("max 度量聚合查询: {}", max); } catch (IOException e) { log.error("max 度量聚合查询异常"); } return max; } /** * avg 统计员工工资平均值 * * @return {@link BigDecimal} */ @Override public BigDecimal aggregationAvg() { BigDecimal avg = BigDecimal.valueOf(-1); AggregationBuilder aggregationBuilder = AggregationBuilders.avg("salary_avg").field("salary"); SearchSourceBuilder sourceBuilder = new SearchSourceBuilder(); sourceBuilder.aggregation(aggregationBuilder); sourceBuilder.size(0); SearchRequest request = new SearchRequest(this.indexName); request.source(sourceBuilder); try { SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT); Aggregations aggs = response.getAggregations(); if (response.status().equals(RestStatus.OK) && aggs != null) { ParsedAvg parsedAvg = aggs.get("salary_avg"); avg = BigDecimal.valueOf(parsedAvg.getValue()); } log.info("avg 度量聚合查询: {}", avg); } catch (IOException e) { log.error("avg 度量聚合查询异常"); } return avg; } /** * sum 统计员工工资总值 * * @return {@link BigDecimal} */ @Override public BigDecimal aggregationSum() { BigDecimal sum = BigDecimal.valueOf(-1); AggregationBuilder aggregationBuilder = AggregationBuilders.sum("salary_sum").field("salary"); SearchSourceBuilder sourceBuilder = new SearchSourceBuilder(); sourceBuilder.aggregation(aggregationBuilder); sourceBuilder.size(0); SearchRequest request = new SearchRequest(this.indexName); request.source(sourceBuilder); try { SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT); Aggregations aggs = response.getAggregations(); if (response.status().equals(RestStatus.OK) && aggs != null) { ParsedSum parsedSum = aggs.get("salary_sum"); sum = BigDecimal.valueOf(parsedSum.getValue()); } log.info("sum 度量聚合查询: {}", sum); } catch (IOException e) { log.error("sum 度量聚合查询异常"); } return sum; } /** * count 统计员工总数 * * @return {@link Long} */ @Override public Long aggregationCount() { long count = -1L; AggregationBuilder aggregationBuilder = AggregationBuilders.count("user_count").field("salary"); SearchSourceBuilder sourceBuilder = new SearchSourceBuilder(); sourceBuilder.aggregation(aggregationBuilder); sourceBuilder.size(0); SearchRequest request = new SearchRequest(this.indexName); request.source(sourceBuilder); try { SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT); Aggregations aggs = response.getAggregations(); if (response.status().equals(RestStatus.OK) && aggs != null) { ParsedValueCount valueCount = aggs.get("user_count"); count = valueCount.getValue(); } log.info("count 度量聚合查询: {}", count); } catch (IOException e) { log.error("count 度量聚合查询异常"); } return count; } /** * percentiles 统计员工工资百分位 * * @return {@link Map}<{@link BigDecimal}, {@link BigDecimal}> */ @Override public Map<BigDecimal, BigDecimal> aggregationPercentiles() { Map<BigDecimal, BigDecimal> map = new HashMap<>(); AggregationBuilder aggregationBuilder = AggregationBuilders.percentiles("salary_percentiles").field("salary"); SearchSourceBuilder sourceBuilder = new SearchSourceBuilder(); sourceBuilder.aggregation(aggregationBuilder); sourceBuilder.size(0); SearchRequest request = new SearchRequest(this.indexName); request.source(sourceBuilder); try { SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT); Aggregations aggs = response.getAggregations(); if (response.status().equals(RestStatus.OK) && aggs != null) { ParsedPercentiles percentiles = aggs.get("salary_percentiles"); for (Percentile percentile : percentiles) { map.put(BigDecimal.valueOf(percentile.getPercent()), BigDecimal.valueOf(percentile.getValue())); } } log.info("percentiles 度量聚合查询: {}", map); } catch (IOException e) { log.error("percentiles 度量聚合查询异常", e); } return map; } }
单元测试:
@Test public void aggregationMin() { BigDecimal min = this.aggsMetricService.aggregationMin(); Assert.assertNotEquals("min 度量聚合查询成功!", -1, min.intValue()); } @Test public void aggregationMax() { BigDecimal max = this.aggsMetricService.aggregationMax(); Assert.assertNotEquals("max 度量聚合查询成功!", -1, max.intValue()); } @Test public void aggregationAvg() { BigDecimal avg = this.aggsMetricService.aggregationAvg(); Assert.assertNotEquals("avg 度量聚合查询成功!", -1, avg.intValue()); } @Test public void aggregationSum() { BigDecimal sum = this.aggsMetricService.aggregationSum(); Assert.assertNotEquals("sum 度量聚合查询成功!", -1, sum.intValue()); } @Test public void aggregationCount() { Long count = this.aggsMetricService.aggregationCount(); Assert.assertNotEquals("count 度量聚合查询成功!", -1, count.intValue()); } @Test public void aggregationPercentiles() { Map<BigDecimal, BigDecimal> map = this.aggsMetricService.aggregationPercentiles(); Assert.assertNotEquals("percentiles 度量聚合查询成功", 0, map.size()); }
// 按岁数进行聚合分桶,统计各个岁数员工的人数: GET /user/_search { "size": 0, "aggs": { "age_bucket": { "terms": { "field": "age", "size": "10" } } } } // 按工资范围进行聚合分桶,统计工资在 200以下、200-500 和 500 以上的员工信息: GET /user/_search { "size": 0, "aggs": { "salary_range_bucket": { "range": { "field": "salary", "ranges": [ { "key": "初级员工", "to": 200 }, { "key": "中级员工", "from": 200, "to": 500 }, { "key": "高级员工", "from": 500 } ] } } } } // 按照时间范围进行分桶,统计 1985-1990 年和 1990-1995 年出生的员工信息: GET /user/_search { "size": 0, "aggs": { "date_range_bucket": { "date_range": { "field": "birthDate", "format": "yyyy", "ranges": [ { "key": "出生日期1990-1995的员工", "from": "1990", "to": "1995" }, { "key": "出生日期1995-2000的员工", "from": "1995", "to": "2000" } ] } } } } // 按工资多少进行聚合分桶,设置统计的最小值为 0,最大值为 1000,区段间隔为 200: GET /user/_search { "size": 0, "aggs": { "salary_histogram": { "histogram": { "field": "salary", "extended_bounds": { "min": 0, "max": 1000 }, "interval": 200 } } } } // 按出生日期进行分桶: GET /user/_search { "size": 0, "aggs": { "birthday_histogram": { "date_histogram": { "format": "yyyy", "field": "birthDate", "interval": "year" } } } }
示例代码:
package com.moon.es.service.impl; import com.moon.es.service.AggsBucketService; import lombok.extern.slf4j.Slf4j; import org.elasticsearch.action.search.SearchRequest; import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.client.RequestOptions; import org.elasticsearch.client.RestHighLevelClient; import org.elasticsearch.rest.RestStatus; import org.elasticsearch.search.aggregations.AggregationBuilder; import org.elasticsearch.search.aggregations.AggregationBuilders; import org.elasticsearch.search.aggregations.Aggregations; import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramInterval; import org.elasticsearch.search.aggregations.bucket.histogram.Histogram; import org.elasticsearch.search.aggregations.bucket.range.Range; import org.elasticsearch.search.aggregations.bucket.terms.Terms; import org.elasticsearch.search.builder.SearchSourceBuilder; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; import java.io.IOException; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; @Slf4j @Service public class AggsBucketServiceImpl implements AggsBucketService { @Autowired private RestHighLevelClient restHighLevelClient; @Value("${elasticsearch.indexName}") private String indexName; /** * 按岁数进行聚合分桶 * * @return {@link Map}<{@link String}, {@link Long}> */ @Override public Map<String, Long> bucketTerms() { Map<String, Long> map = new LinkedHashMap<>(); AggregationBuilder aggregationBuilder = AggregationBuilders.terms("age_bucket").field("age"); SearchSourceBuilder sourceBuilder = new SearchSourceBuilder(); sourceBuilder.size(0); sourceBuilder.aggregation(aggregationBuilder); SearchRequest request = new SearchRequest(this.indexName); request.source(sourceBuilder); try { SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT); Aggregations aggs = response.getAggregations(); if (response.status().equals(RestStatus.OK) && aggs != null) { // 分桶 Terms termsBucket = aggs.get("age_bucket"); List<? extends Terms.Bucket> buckets = termsBucket.getBuckets(); for (Terms.Bucket bucket : buckets) { map.put(bucket.getKeyAsString(), bucket.getDocCount()); } } log.info("term 桶聚合查询: {}", map); } catch (IOException e) { log.error("term 桶聚合查询异常"); } return map; } /** * 按工资范围进行聚合分桶 * * @return {@link Map}<{@link String}, {@link Long}> */ @Override public Map<String, Long> bucketRange() { Map<String, Long> map = new LinkedHashMap<>(); AggregationBuilder aggregationBuilder = AggregationBuilders.range("salary_range_bucket") .field("salary") .addUnboundedTo("初级员工", 200) .addRange("中级员工", 200, 500) .addUnboundedFrom("高级员工", 500); SearchSourceBuilder sourceBuilder = new SearchSourceBuilder(); sourceBuilder.size(0); sourceBuilder.aggregation(aggregationBuilder); SearchRequest request = new SearchRequest(this.indexName); request.source(sourceBuilder); try { SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT); Aggregations aggs = response.getAggregations(); if (response.status().equals(RestStatus.OK) && aggs != null) { Range rangeBucket = aggs.get("salary_range_bucket"); List<? extends Range.Bucket> buckets = rangeBucket.getBuckets(); for (Range.Bucket bucket : buckets) { map.put(bucket.getKeyAsString(), bucket.getDocCount()); } } log.info("range 桶聚合查询: {}", map); } catch (IOException e) { log.error("range 桶聚合查询异常"); } return map; } /** * 按照时间范围进行分桶 * * @return {@link Map}<{@link String}, {@link Long}> */ @Override public Map<String, Long> bucketDateRange() { Map<String, Long> map = new LinkedHashMap<>(); AggregationBuilder aggregationBuilder = AggregationBuilders.dateRange("date_range_bucket") .field("birthDate") .format("yyyy") .addRange("1990-1995", "1990", "1995") .addRange("1995-2000", "1995", "2000"); SearchSourceBuilder sourceBuilder = new SearchSourceBuilder(); sourceBuilder.size(0); sourceBuilder.aggregation(aggregationBuilder); SearchRequest request = new SearchRequest(this.indexName); request.source(sourceBuilder); try { SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT); Aggregations aggs = response.getAggregations(); if (response.status().equals(RestStatus.OK) && aggs != null) { Range dateRangeBucket = aggs.get("date_range_bucket"); List<? extends Range.Bucket> buckets = dateRangeBucket.getBuckets(); for (Range.Bucket bucket : buckets) { map.put(bucket.getKeyAsString(), bucket.getDocCount()); } } log.info("date_range 桶聚合查询: {}", map); } catch (IOException e) { log.error("date_range 桶聚合查询异常"); } return map; } /** * 按工资多少进行聚合分桶 * * @return {@link Map}<{@link String}, {@link Long}> */ @Override public Map<String, Long> bucketHistogram() { Map<String, Long> map = new LinkedHashMap<>(); AggregationBuilder aggregationBuilder = AggregationBuilders.histogram("salary_histogram") .field("salary") .extendedBounds(0, 1000) .interval(200); SearchSourceBuilder sourceBuilder = new SearchSourceBuilder(); sourceBuilder.size(0); sourceBuilder.aggregation(aggregationBuilder); SearchRequest request = new SearchRequest(this.indexName); request.source(sourceBuilder); try { SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT); Aggregations aggs = response.getAggregations(); if (response.status().equals(RestStatus.OK) && aggs != null) { Histogram histogramBucket = aggs.get("salary_histogram"); List<? extends Histogram.Bucket> buckets = histogramBucket.getBuckets(); for (Histogram.Bucket bucket : buckets) { map.put(bucket.getKeyAsString(), bucket.getDocCount()); } } log.info("histogram 桶聚合查询: {}", map); } catch (IOException e) { log.error("histogram 桶聚合查询异常"); } return map; } /** * 按出生日期进行分桶 * * @return {@link Map}<{@link String}, {@link Long}> */ @Override public Map<String, Long> bucketDateHistogram() { Map<String, Long> map = new LinkedHashMap<>(); AggregationBuilder aggregationBuilder = AggregationBuilders.dateHistogram("birthdate_histogram") .field("birthDate") .format("yyyy") .calendarInterval(DateHistogramInterval.YEAR); SearchSourceBuilder sourceBuilder = new SearchSourceBuilder(); sourceBuilder.size(0); sourceBuilder.aggregation(aggregationBuilder); SearchRequest request = new SearchRequest(this.indexName); request.source(sourceBuilder); try { SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT); Aggregations aggs = response.getAggregations(); if (response.status().equals(RestStatus.OK) && aggs != null) { Histogram dateHistogramBucket = aggs.get("birthdate_histogram"); List<? extends Histogram.Bucket> buckets = dateHistogramBucket.getBuckets(); for (Histogram.Bucket bucket : buckets) { map.put(bucket.getKeyAsString(), bucket.getDocCount()); } } log.info("date_histogram 桶聚合查询: {}", map); } catch (IOException e) { log.error("date_histogram 桶聚合查询异常"); } return map; } }
单元测试:
@Test public void bucketTerms() { Map<String, Long> map = this.aggsBucketService.bucketTerms(); Assert.assertNotEquals("term 桶聚合查询成功!", 0, map.size()); } @Test public void bucketRange() { Map<String, Long> map = this.aggsBucketService.bucketRange(); Assert.assertNotEquals("range 桶聚合查询成功!", 0, map.size()); } @Test public void bucketDateRange() { Map<String, Long> map = this.aggsBucketService.bucketDateRange(); Assert.assertNotEquals("date_range 桶聚合查询成功!", 0, map.size()); } @Test public void bucketHistogram() { Map<String, Long> map = this.aggsBucketService.bucketHistogram(); Assert.assertNotEquals("histogram 桶聚合查询成功!", 0, map.size()); } @Test public void bucketDateHistogram() { Map<String, Long> map = this.aggsBucketService.bucketDateHistogram(); Assert.assertNotEquals("date_histogram 桶聚合查询成功!", 0, map.size()); }
GET /user/_search { "size": 0, "aggs": { "salary_bucket": { "terms": { "field": "age", "size": "10" }, "aggs": { "salary_max_user": { "top_hits": { "size": 1, "sort": [ { "salary": { "order": "desc" } } ] } } } } } }
示例代码:
package com.moon.es.service.impl; import com.alibaba.fastjson.JSON; import com.moon.es.entity.User; import com.moon.es.service.AggsBucketMetricService; import lombok.extern.slf4j.Slf4j; import org.elasticsearch.action.search.SearchRequest; import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.client.RequestOptions; import org.elasticsearch.client.RestHighLevelClient; import org.elasticsearch.rest.RestStatus; import org.elasticsearch.search.SearchHit; import org.elasticsearch.search.aggregations.AggregationBuilder; import org.elasticsearch.search.aggregations.AggregationBuilders; import org.elasticsearch.search.aggregations.Aggregations; import org.elasticsearch.search.aggregations.bucket.terms.Terms; import org.elasticsearch.search.aggregations.metrics.ParsedTopHits; import org.elasticsearch.search.builder.SearchSourceBuilder; import org.elasticsearch.search.sort.SortOrder; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; import java.io.IOException; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; @Slf4j @Service public class AggsBucketMetricServiceImpl implements AggsBucketMetricService { @Autowired private RestHighLevelClient restHighLevelClient; @Value("${elasticsearch.indexName}") private String indexName; /** * topHits 按岁数分桶、然后统计每个员工工资最高值 * * @return {@link Map}<{@link String}, {@link User}> */ @Override public Map<String, User> aggregationTopHits() { Map<String, User> map = new LinkedHashMap<>(); AggregationBuilder salaryBucket = AggregationBuilders.terms("salary_bucket") .field("age") .size(10); AggregationBuilder salaryMaxUser = AggregationBuilders.topHits("salary_max_user") .size(1) .sort("salary", SortOrder.DESC); salaryBucket.subAggregation(salaryMaxUser); SearchSourceBuilder sourceBuilder = new SearchSourceBuilder(); sourceBuilder.size(0); sourceBuilder.aggregation(salaryBucket); SearchRequest request = new SearchRequest(this.indexName); request.source(sourceBuilder); try { SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT); Aggregations aggs = response.getAggregations(); if (response.status().equals(RestStatus.OK) && aggs != null) { Terms termsBucket = aggs.get("salary_bucket"); List<? extends Terms.Bucket> buckets = termsBucket.getBuckets(); for (Terms.Bucket bucket : buckets) { ParsedTopHits topHits = bucket.getAggregations().get("salary_max_user"); for (SearchHit hit : topHits.getHits()) map.put(bucket.getKeyAsString() + "=>" + bucket.getDocCount(), JSON.parseObject(hit.getSourceAsString(), User.class)); } } log.error("top_hits 聚合查询: {}", map); } catch (IOException e) { log.error("top_hits 聚合查询异常", e); } return map; } }
单元测试:
@Test
public void aggregationTopHits() {
Map<String, User> map = this.aggsBucketMetricService.aggregationTopHits();
Assert.assertNotEquals("top_hits 聚合查询成功!", 0, map.size());
}
gram 桶聚合查询成功!", 0, map.size());
}
### Metric 与 Bucket 聚合 ```json GET /user/_search { "size": 0, "aggs": { "salary_bucket": { "terms": { "field": "age", "size": "10" }, "aggs": { "salary_max_user": { "top_hits": { "size": 1, "sort": [ { "salary": { "order": "desc" } } ] } } } } } }
示例代码:
package com.moon.es.service.impl; import com.alibaba.fastjson.JSON; import com.moon.es.entity.User; import com.moon.es.service.AggsBucketMetricService; import lombok.extern.slf4j.Slf4j; import org.elasticsearch.action.search.SearchRequest; import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.client.RequestOptions; import org.elasticsearch.client.RestHighLevelClient; import org.elasticsearch.rest.RestStatus; import org.elasticsearch.search.SearchHit; import org.elasticsearch.search.aggregations.AggregationBuilder; import org.elasticsearch.search.aggregations.AggregationBuilders; import org.elasticsearch.search.aggregations.Aggregations; import org.elasticsearch.search.aggregations.bucket.terms.Terms; import org.elasticsearch.search.aggregations.metrics.ParsedTopHits; import org.elasticsearch.search.builder.SearchSourceBuilder; import org.elasticsearch.search.sort.SortOrder; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; import java.io.IOException; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; @Slf4j @Service public class AggsBucketMetricServiceImpl implements AggsBucketMetricService { @Autowired private RestHighLevelClient restHighLevelClient; @Value("${elasticsearch.indexName}") private String indexName; /** * topHits 按岁数分桶、然后统计每个员工工资最高值 * * @return {@link Map}<{@link String}, {@link User}> */ @Override public Map<String, User> aggregationTopHits() { Map<String, User> map = new LinkedHashMap<>(); AggregationBuilder salaryBucket = AggregationBuilders.terms("salary_bucket") .field("age") .size(10); AggregationBuilder salaryMaxUser = AggregationBuilders.topHits("salary_max_user") .size(1) .sort("salary", SortOrder.DESC); salaryBucket.subAggregation(salaryMaxUser); SearchSourceBuilder sourceBuilder = new SearchSourceBuilder(); sourceBuilder.size(0); sourceBuilder.aggregation(salaryBucket); SearchRequest request = new SearchRequest(this.indexName); request.source(sourceBuilder); try { SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT); Aggregations aggs = response.getAggregations(); if (response.status().equals(RestStatus.OK) && aggs != null) { Terms termsBucket = aggs.get("salary_bucket"); List<? extends Terms.Bucket> buckets = termsBucket.getBuckets(); for (Terms.Bucket bucket : buckets) { ParsedTopHits topHits = bucket.getAggregations().get("salary_max_user"); for (SearchHit hit : topHits.getHits()) map.put(bucket.getKeyAsString() + "=>" + bucket.getDocCount(), JSON.parseObject(hit.getSourceAsString(), User.class)); } } log.info("top_hits 聚合查询: {}", map); } catch (IOException e) { log.error("top_hits 聚合查询异常", e); } return map; } }
单元测试:
@Test
public void aggregationTopHits() {
Map<String, User> map = this.aggsBucketMetricService.aggregationTopHits();
Assert.assertNotEquals("top_hits 聚合查询成功!", 0, map.size());
}
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。