Elasticsearch高级

文章目录

一.数据聚合
二.RestAPI实现聚合
三.ES自动补全(联想)
四.数据同步
五.elasticsearch集群

一.数据聚合

在ES中的数据聚合（aggregations）可以近似看做成mysql中的groupby分组,聚合可以实现对文档数据的统计、分析、运算,常见的聚合的分类有以下几种:

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

1.DSL实现Bucket(桶)聚合

语法:

GET /hotel/_search
{"size": 0,  // 设置size为0，结果中不包含文档，只包含聚合结果"aggs": { // 定义聚合(带有"s"证明可以定义多个聚合)"xxx": { //给聚合起个名字"terms": { // 聚合的类型(不能为text)"field": "xxx", // 参与聚合的字段"size": xxx // 希望获取的聚合结果数量}}}
}

默认情况下，Bucket聚合会统计Bucket内的文档数量，记为_count，并且按照_count降序排序。修改结果排序方式:

GET /hotel/_search
{"size": 0, "aggs": {"xxx": {"terms": {"field": "xxx","order": {"_count": "asc" // 按照_count升序排列},"size": 20}}}
}

默认情况下，Bucket聚合是对索引库的所有文档做聚合，我们可以限定要聚合的文档范围，只要添加query条件即可

示例:

GET /hotel/_search
{"query": {"range": {"price": {"lte": 200 // 只对200元以下的文档聚合}}}, "size": 0, "aggs": {"brandAgg": {"terms": {"field": "brand","size": 20}}}
}

2.DSL实现Metrics 聚合

例如，我们要求获取每个品牌的用户评分的min、max、avg等值.
我们可以利用stats聚合：

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

二.RestAPI实现聚合

1.发送聚合DSL语句(其实本质上就是一层一层按照DSL语句组成结构调用API,十分易懂也很简单):

在这里插入图片描述
2.聚合结果解析

java代码:

@Testvoid aggregation() throws IOException {SearchRequest request=new SearchRequest("hotel");request.source().size(0);request.source().aggregation(AggregationBuilders.terms("brandAgg").field("brand").size(10));SearchResponse response = client.search(request, RequestOptions.DEFAULT);Aggregations aggregations = response.getAggregations();Terms terms=aggregations.get("brandAgg");List<? extends Terms.Bucket> buckets = terms.getBuckets();for (Terms.Bucket bucket : buckets) {String key = bucket.getKeyAsString();System.out.println(key);}}

三.ES自动补全(联想)

用途:当用户在搜索框输入字符时，我们应该提示出与该字符(汉字,英文)有关的搜索项

1.拼音分词

要实现根据字母做补全，就必须对文档按照拼音分词。在GitHub上恰好有elasticsearch的拼音分词插件。

下载地址:https://github.com/medcl/elasticsearch-analysis-pinyin

自定义分词器

elasticsearch中分词器（analyzer）的组成包含三部分：

character filters：在tokenizer之前对文本进行处理。例如删除字符、替换字符
tokenizer：将文本按照一定的规则切割成词条（term）。例如keyword，就是不分词；还有ik_smart
tokenizer filter：将tokenizer输出的词条做进一步处理。例如大小写转换、同义词处理、拼音处理等

在这里插入图片描述

我们可以在创建索引库时，通过settings来配置自定义的analyzer（分词器）：

PUT /test
{"settings": {"analysis": {"analyzer": { // 自定义分词器"my_analyzer": {  // 分词器名称"tokenizer": "ik_max_word","filter": "py"}},"filter": { // 自定义tokenizer filter"py": { // 过滤器名称"type": "pinyin", // 过滤器类型，这里是pinyin"keep_full_pinyin": false,"keep_joined_full_pinyin": true,"keep_original": true,"limit_first_letter_length": 16,"remove_duplicated_term": true,"none_chinese_pinyin_tokenize": false}}}}
}

拼音分词器适合在创建倒排索引的时候使用，但不能在搜索的时候使用(原因是搜索时会把文本转换为多个拼音和分词(如果添加了tokenizer分词的话),按照拼音搜索就会和倒排索引库里面的拼音匹配(包括拼音一致但是词义不一致的词),这就会导致搜索产生歧义)。

因此字段在创建倒排索引时应该用my_analyzer分词器；字段在搜索时应该使用ik_smart分词器(使用ik_smart时识别不了拼音,es会默认使用创建索引库时指定的分词器)

PUT /test
{"settings": {"analysis": {"analyzer": {"my_analyzer": {"tokenizer": "ik_max_word", "filter": "py"}},"filter": {"py": { ... }}}},"mappings": {"properties": {"name": {"type": "text","analyzer": "my_analyzer","search_analyzer": "ik_smart"}}}
}

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

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

在这里插入图片描述

查询语法如下：

在这里插入图片描述
3.RestAPI实现自动补全

请求参数构造的API：
在这里插入图片描述
结果解析：

案例:实现酒店搜索页面输入框的自动补全

业务层代码:

 public List<String> getSuggestions(String prefix) {try {SearchRequest request=new SearchRequest("hotel");request.source().suggest(new SuggestBuilder().addSuggestion("suggestions",SuggestBuilders.completionSuggestion("suggestion").prefix(prefix).skipDuplicates(true).size(10)));SearchResponse response = client.search(request, RequestOptions.DEFAULT);Suggest suggest = response.getSuggest();CompletionSuggestion completionSuggestion = suggest.getSuggestion("suggestions");List<CompletionSuggestion.Entry.Option> options = completionSuggestion.getOptions();List<String> suggestionList=new ArrayList<>(options.size());for (CompletionSuggestion.Entry.Option option : options) {String text = option.getText().toString();suggestionList.add(text);}return suggestionList;} catch (IOException e) {throw new RuntimeException(e);}}

四.数据同步

数据同步问题分析

elasticsearch中的酒店数据来自于mysql数据库，因此mysql数据发生改变时，elasticsearch也必须跟着改变，这个就是elasticsearch与mysql之间的数据同步。

方案一：同步调用(不推荐使用)

在这里插入图片描述

方案二：异步通知(推荐使用)

在这里插入图片描述

方案三：监听binlog(推荐使用)

在这里插入图片描述

总结:

方式一：同步调用

优点：实现简单，粗暴
缺点：业务耦合度高

方式二：异步通知

优点：低耦合，实现难度一般
缺点：依赖mq的可靠性

方式三：监听binlog

优点：完全解除服务间耦合
缺点：开启binlog增加数据库负担、实现复杂度高

案例:利用MQ实现mysql与elasticsearch数据同步

1.在hotel-demo引入amqp依赖

		<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency>

2.编写RabbitMQ的连接信息(yaml文件中,省略)

3.编写常量RabbitMQ的交换机,队列,Routing_Key的值

package cn.itcast.hotel.constants;public class RabbitMQConstants {/*** 交换机*/public static final String HOTEL_EXCHANGE="hotel.topic";/*** 新增或修改业务队列*/public static final String HOTEL_INSERT_QUEUE="hotel.insert.queue";/*** 删除业务队列*/public static final String HOTEL_DELETE_QUEUE="hotel.delete.queue";/*** 新增或修改业务ROUTING_KEY*/public static final String HOTEL_INSERT_KEY="hotel.insert";/*** 修改业务的ROUTING_KEY*/public static final String HOTEL_DELETE_KEY="hotel.delete";
}

4.定义交换机,队列,绑定交换机和队列,以及声明Routing_Key

package cn.itcast.hotel.mq;import cn.itcast.hotel.constants.RabbitMQConstants;
import cn.itcast.hotel.service.IHotelService;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;@Component
public class HotelListener {@Autowiredprivate IHotelService iHotelService;/*** 监听新增或修改的消息* @param id*/@RabbitListener(bindings = @QueueBinding(value = @Queue(name = RabbitMQConstants.HOTEL_INSERT_QUEUE),exchange = @Exchange(name = RabbitMQConstants.HOTEL_EXCHANGE),key = RabbitMQConstants.HOTEL_INSERT_KEY))public void listenHotelInsertOrUpdate(Long id){iHotelService.insertById(id);}/*** 监听删除的消息* @param id*/@RabbitListener(bindings = @QueueBinding(value = @Queue(name = RabbitMQConstants.HOTEL_DELETE_QUEUE),exchange = @Exchange(name = RabbitMQConstants.HOTEL_EXCHANGE),key = RabbitMQConstants.HOTEL_DELETE_KEY))public void listenHotelDelete(Long id){iHotelService.deleteById(id);}
}

5.在hotel-admin重复操作1,2,3

6.当执行增删改操作,发送消息到交换机中

	@PostMappingpublic void saveHotel(@RequestBody Hotel hotel) {hotelService.save(hotel);rabbitTemplate.convertAndSend(RabbitMQConstants.HOTEL_EXCHANGE, RabbitMQConstants.HOTEL_INSERT_KEY, hotel.getId());}@PutMapping()public void updateById(@RequestBody Hotel hotel) {if (hotel.getId() == null) {throw new InvalidParameterException("id不能为空");}hotelService.updateById(hotel);rabbitTemplate.convertAndSend(RabbitMQConstants.HOTEL_EXCHANGE, RabbitMQConstants.HOTEL_INSERT_KEY, hotel.getId());}@DeleteMapping("/{id}")public void deleteById(@PathVariable("id") Long id) {hotelService.removeById(id);rabbitTemplate.convertAndSend(RabbitMQConstants.HOTEL_EXCHANGE, RabbitMQConstants.HOTEL_DELETE_KEY, id);}

7.在hotel-demo处理消息(业务层实现)

public void insertById(Long id) {try {Hotel hotel = getById(id);HotelDoc hotelDoc = new HotelDoc(hotel);IndexRequest request = new IndexRequest("hotel").id(hotel.getId().toString());request.source(JSON.toJSONString(hotelDoc), XContentType.JSON);client.index(request, RequestOptions.DEFAULT);} catch (IOException e) {throw new RuntimeException(e);}}public void deleteById(Long id) {try {DeleteRequest request=new DeleteRequest("hotel").id(id.toString());client.delete(request, RequestOptions.DEFAULT);} catch (IOException e) {throw new RuntimeException(e);}}