Elasticsearch集群搭建学习

Elasticsearch集群聚合、集群搭建

- RestClient
- - 查询所有
  - 高亮
  - 算分控制
- 数据聚合
- - DSL实现Bucket聚合
  - DSL实现Metrics聚合
  - RestAPI实现聚合
- 拼音分词器
- - 如何使用拼音分词器？
  - 如何自定义分词器？
  - 拼音分词器注意事项？
- 自动补全
- 数据同步
- 集群搭建
- - ES集群结构
  - 创建es集群
  - 集群状态监控
  - 创建索引库
  - - 1）利用kibana的DevTools创建索引库
    - 2）利用cerebro创建索引库
  - 查看分片效果
  - ES集群中的节点角色
  - ES脑裂
  - ES集群的分布式存储
  - ES集群的分布式查询
  - ES集群的故障转移

RestClient

查询所有

void testMatchAll() throws IOException {//1.准备RequestSearchRequest request = new SearchRequest("hotel");//2.准备DSLrequest.source().query(QueryBuilders.matchAllQuery());//3.发送请求SearchResponse response = restClient.search(requ est , RequestOptions.DEFAULT);System.out.println(response);}

高亮

高亮API包括请求DSL构建和结果解析两部分。

构建：

request.source().highlighter(new HghlightBuilder().field("name")//是否需要与字段匹配.requireFieldMatch(false))

@Testpublic void testHighLight() throws IOException {//1 准备RequestSearchRequest request = new SearchRequest("hotel");//2 准备DSLrequest.source().query(QueryBuilders.matchQuery("name" , "如家"));//2.2 高亮request.source().highlighter(new HighlightBuilder().field("name").requireFieldMatch(false));//3 发送请求SearchResponse response = restClient.search(request , RequestOptions.DEFAULT);//4 解析响应SearchHits searchHits = response.getHits();//5 获取总条数long total = searchHits.getTotalHits().value;//4.2 文档数组SearchHit[] hits = searchHits.getHits();//4.3 遍历for(SearchHit hit : hits){//获取文档sourceString json = hit.getSourceAsString();//反序列化HotelDoc hotelDoc = JSON.parseObject(json , HotelDoc.class);//获取高亮结果Map<String , HighlightField> highlightFields = hit.getHighlightFields();//根据字段名获取高亮结果HighlightField highlightField = hightlightFields.get("name");//获取高亮值String name = highlightField.getFragments()[0].string();//覆盖非高亮结果hotelDoc.setName(name);}}

所有搜索DSL的构建，记住一个API：
- SearchRequest的source()方法。
高亮结果解析是参考JSON结果，逐层解析

算分控制

FunctionScoreQueryBuilder functionScoreQuery = QueryBuilders.functionScoreQuery(//原始查询，相关性算分的查询boolQuery ,new FunctionSocreQueryBuilder.FilterFunctionBuilder[]{new FunctionScoreQueryBuilder.FilterFunctionBuilder(//过滤条件QueryBuilders.termQuery("isAd" , true),//算分函数ScoreFuntionBuilders.weightFactorFuntion(10))}
);

数据聚合

聚合的分类

聚合（aggregations）可以实现对文档数据的统计、分析、运算。聚合常见的有三类：

桶（Bucket）聚合：用来对文档做分组
- TermAggregation:按照文档字段值分组
- Date Histogram:按照日期阶梯分组，例如一周为一组，或者一月为一组
度量（Metric）聚合：用以计算一些值，比如：最大值、最小值、平均值等
- Avg：求平均值
- Max：求最大值
- Min：求最小值
- Stats：同时求max、min、avg、sum等
管道（pipeline）聚合：其他聚合的结果为基础做聚合

什么是聚合？

聚合是对文档数据的统计、分析、计算

聚合的常见种类有哪些？

Bucket：对文档数据分组，并统计每组数量
Metric：对文档数据做计算，例如avg
Pipeline：基于其他聚合结果再做聚合

参与聚合的字段类型必须是：

keyword
数值
日期
布尔

DSL实现Bucket聚合

GET /hotel/_search
{"size":0, // 设置size为0，结果中不包含文档，只包含聚合结果"aggs":{ // 定义聚合"brandAgg":{ // 给聚合起个名字"terms":{ // 聚合的类型，按照品牌值聚合，所以选择term"field":"brand", // 参与聚合的字段"size":20 // 希望获取的聚合结果数量}}}
}

aggs代表聚合，与query同级，此时query的作用是？

限定聚合的文档范围

聚合必须的三要素：

聚合名称
聚合类型
聚合手段

聚合可配置属性有：

size：指定聚合结果数量
order：指定聚合结果排序方式
field：指定聚合字段

DSL实现Metrics聚合

例如，我们要求获取每个品牌的用户评分的min、max、avg等值

GET /hotel/_search
{"size":0,"aggs":{"brandAgg":{"terms":{"field":"brand","size":20},"aggs":{// 是brands聚合的子聚合，也就是分组后对每组分别计算"score_status":{ // 聚合名称"stats":{ // 聚合类型，这里stats可以计算min、max、avg等"field":"score"// 聚合字段，这里可以是score}}}}}
}

RestAPI实现聚合

@Testpublic void testAggregation() throws IOException {// 准备RequestSearchRequest request = new SearchRequest("hotel");// 准备DSL// 设置sizerequest.source().size(0);// 聚合request.source().aggregation(AggregationBuilders.terms("brandAgg").size(20).field("brand"));//发出请求SearchResponse response = restClient.search(request, RequestOptions.DEFAULT);//解析聚合结果Aggregations aggregations = response.getAggregations();//根据名称获取聚合结果Terms brandTerms = aggregations.get("brand_agg");//获取桶List<? extends Terms.Bucket> buckets = brandTerms.getBuckets();for (Terms.Bucket bucket : buckets) {//获取key，也就是品牌信息String brandName = bucket.getKeyAsString();System.out.println(brandName);}}

实现对品牌、城市、星级的聚合

@Testpublic Map<String , List<String>> testMap() throws IOException {// 准备RequestSearchRequest request = new SearchRequest("hotel");// 准备DSL// 设置sizerequest.source().size(0);// 聚合request.source().aggregation(AggregationBuilders.terms("brandAgg").size(20).field("brand"));request.source().aggregation(AggregationBuilders.terms("cityAgg").size(20).field("city"));request.source().aggregation(AggregationBuilders.terms("starAgg").size(20).field("starName"));//发出请求SearchResponse response = restClient.search(request, RequestOptions.DEFAULT);//解析聚合结果Aggregations aggregations = response.getAggregations();Map<String , List<String>> result = new HashMap<>();result.put("brandAgg" , getAggByName("brandAgg" , aggregations));result.put("cityAgg" , getAggByName("cityAgg" , aggregations));result.put("starAgg" , getAggByName("starAgg" , aggregations));return result;}private List<String> getAggByName(String name , Aggregations aggregations){List<String> result = new ArrayList<>();Terms terms = aggregations.get(name);for (Terms.Bucket bucket : terms.getBuckets()) {String keyAsString = bucket.getKeyAsString();result.add(keyAsString);}return result;}

拼音分词器

如何使用拼音分词器？

下载pinyin分词器

解压并放到elasticsearch的plugin目录

重启

如何自定义分词器？

创建索引库时，在settings中配置，可以包含三部分

character filter

tokenizer

filter

拼音分词器注意事项？

为了避免搜索到同音字，搜索时不要使用拼音分词器

自动补全

elasticsearch提供了Completion Suggester查询来实现自动补全功能。这个查询会匹配以用户输入内容开头的词条并返回。为了提高补全查询的效率，对于文档中字段的类型有一些约束：

参与补全查询的字段必须是completion类型
字段的内容一般是用来补全的多个词条形成的数组

// 自动补全查询
GET /test2/_search
{"suggest":{"titleSuggest":{"text":"s","completion":{"field":"title","skip_duplicates":true,"size":10}}}
}

自动补全对字段的要求：

类型是completion类型
字段值是多词条的数组

@Test
public void testSuggest() throws IOException {// 准备RequestSearchRequest request = new SearchRequest("hotel");// 准备DSLrequest.source().suggest(new SuggestBuilder().addSuggestion("suggestions" , SuggestBuilders.completionSuggestion("suggestion").prefix("h").skipDuplicates(true).size(10)));// 发起请求SearchResponse response = client.search(request,RequestOptions.DEFAULT);// 解析结果Suggest suggest = response.getSuggest();// 根据名称获取补全结果CompletionSuggestion suggestion = suggest.getSuggestion("hotelSuggestion");//获取options并遍历for(CompletionSuggestion.Entry.Option option : suggestion.getOptions()){String text = option.getText().string();System.out.println(text);}
}

数据同步

同步调用
- 优点：实现简单，粗暴
- 缺点：业务耦合度高
异步通知
- 优点：低耦合，实现难度一般
- 缺点：依赖mq的可靠性
监听binlog
- 优点：完全解除服务间耦合
- 缺点：开启binlog增加数据库负担、实现复杂度高

异步通知数据同步步骤：

定义config文件，声明队列和交换机bean，并绑定队列与交换机
在增加删除修改接口中发送mq消息到指定的增删改队列
定义监听器，监听mq消息并修改es文档

集群搭建

ES集群结构

单机的es做数据存储，必然面临两个问题：海量数据存储问题，单点故障问题

海量数据存储问题：将索引库从逻辑上拆分N个分片（shard），存储到多个节点
单点故障问题：将分片数据在不同节点备份（replica）

创建es集群

首先编写一个docker-compose.yml文件，内容如下：

version: '2.2'
services:es01:image: elasticsearch:7.12.1container_name: es01environment:- node.name=es01- cluster.name=es-docker-cluster- discovery.seed_hosts=es02,es03- cluster.initial_master_nodes=es01,es02,es03- "ES_JAVA_OPTS=-Xms512m -Xmx512m"volumes:- data01:/usr/share/elasticsearch/dataports:- 9200:9200networks:- elastices02:image: elasticsearch:7.12.1container_name: es02environment:- node.name=es02- cluster.name=es-docker-cluster- discovery.seed_hosts=es01,es03- cluster.initial_master_nodes=es01,es02,es03- "ES_JAVA_OPTS=-Xms512m -Xmx512m"volumes:- data02:/usr/share/elasticsearch/dataports:- 9201:9200networks:- elastices03:image: elasticsearch:7.12.1container_name: es03environment:- node.name=es03- cluster.name=es-docker-cluster- discovery.seed_hosts=es01,es02- cluster.initial_master_nodes=es01,es02,es03- "ES_JAVA_OPTS=-Xms512m -Xmx512m"volumes:- data03:/usr/share/elasticsearch/datanetworks:- elasticports:- 9202:9200
volumes:data01:driver: localdata02:driver: localdata03:driver: localnetworks:elastic:driver: bridge

es运行需要修改一些linux系统权限，修改/etc/sysctl.conf文件

vi /etc/sysctl.conf

添加下面的内容：

vm.max_map_count=262144

然后执行命令，让配置生效：

sysctl -p

通过docker-compose启动集群：

docker-compose up -d

集群状态监控

kibana可以监控es集群，不过新版本需要依赖es的x-pack 功能，配置比较复杂。

这里推荐使用cerebro来监控es集群状态，官方网址：https://github.com/lmenezes/cerebro

双击的cerebro.bat文件即可启动服务。

访问http://localhost:9000 即可进入管理界面：

在这里插入图片描述

输入你的elasticsearch的任意节点的地址和端口，点击connect即可：

在这里插入图片描述

绿色的条，代表集群处于绿色（健康状态）。

创建索引库

1）利用kibana的DevTools创建索引库

在DevTools中输入指令：

PUT /itcast
{"settings": {"number_of_shards": 3, // 分片数量"number_of_replicas": 1 // 副本数量},"mappings": {"properties": {// mapping映射定义 ...}}
}

2）利用cerebro创建索引库

利用cerebro还可以创建索引库：

在这里插入图片描述

填写索引库信息：

在这里插入图片描述

点击右下角的create按钮：

在这里插入图片描述

查看分片效果

回到首页，即可查看索引库分片效果：

在这里插入图片描述

ES集群中的节点角色

elasticsearch中集群节点有不同的职责划分：

在这里插入图片描述

ES脑裂

默认情况下，每个节点都是master eligible节点，因此一旦master节点宕机，其他候选节点会选举一个称为主节点。当主节点与其他节点网络故障时，可能发生脑裂问题。

为了避免脑裂，需要要求选票超过（eligible节点数量+1）/2才能当选为主，因此eligible节点数量最好是奇数。对应配置项是discovery.zen.minimux_master_nodes,在es7.0以后，已经成为默认配置，因此一般不会发生脑裂问题。

master eligible结点的作用？
- 参与集群选主
- 主节点可以管理集群状态、管理分片信息、处理创建和删除索引库的请求
data结点的作用？
- 数据的CRUD
coordinator结点的作用？
- 路由请求到其他节点
- 合并查询到的结果，返回给用户

ES集群的分布式存储

当新增文档时，应该保存到不同的分片，保证数据均衡，那么coordinating node如何确定数据该存储到哪个分片呢？elasticsearch会通过hash算法来计算文档应该存储到哪个分片：

shard = hash(_routing) % number_of_shards

说明：

_routing默认是文档的id
算法与分片数量有关，因此索引库一旦创建，分片数量不能修改！

ES集群的分布式查询

elasticsearch的查询分为两个阶段：

scatter phase:分散阶段，coordinating node会把请求分发到每一个分片
gather phase:聚集阶段，coordinating node汇总data node的搜索结果，并处理为最终结果集返回给用户

ES集群的故障转移

集群的master节点会监控集群中的节点状态，如果发现有节点宕机，会立即将宕机节点的分片数据迁移到其他节点，确保数据安全，这个叫做故障转移