当前位置：网站首页>ES 数据聚合、数据同步、集群

ES 数据聚合、数据同步、集群

2022-08-04 21:49:00 【从零开始的JAVA世界】

1.数据聚合

聚合（aggregations ）可以让我们极其方便的实现对数据的统计、分析、运算。例如：

什么品牌的手机最受欢迎？
这些手机的平均价格、最高价格、最低价格？
这些手机每月的销售情况如何？

实现这些统计功能的比数据库的sql要方便的多，而且查询速度非常快，可以实现近实时搜索效果。

1.1.聚合的种类

聚合常见的有三类：

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

注意:参加聚合的字段必须是keyword、日期、数值、布尔类型，不做分词操作的字段。

1.2.DSL实现聚合

现在，我们要统计所有数据中的酒店品牌有几种，其实就是按照品牌对数据分组。此时可以根据酒店品牌的名称做聚合，也就是Bucket聚合。

1.2.1.Bucket聚合语法

语法如下：

GET /hotel/_search
{
    
  "aggs": {
     // 定义聚合
    "brandAgg": {
      //给聚合起个名字
      "terms": {
       // 聚合的类型，按照品牌值聚合，所以选择term
        "field": "brand", // 参与聚合的字段
        "size": 20 // 显示的聚合结果数量
      }
    }
  }
}

1.2.2.聚合结果排序

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

我们可以指定order属性，自定义聚合的排序方式：

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

1.2.3.限定聚合范围

默认情况下，Bucket聚合是对索引库的所有文档做聚合，但真实场景下，用户会输入搜索条件，因此聚合必须是对搜索结果聚合。那么聚合必须添加限定条件。

我们可以限定要聚合的文档范围，只要添加query条件即可：

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

1.2.4.Metric聚合语法

桶内聚合，使用聚合函数

语法如下：

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

1.2.5.小结

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

限定聚合的的文档范围

聚合必须的三要素：

聚合名称
聚合类型
聚合字段

聚合可配置属性有：

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

1.3.RestAPI实现聚合

    @Test
    public void testAggs() throws IOException {
    
        //创建搜索请求对象
        SearchRequest searchRequest = new SearchRequest("hotel");
        //查询条件
        searchRequest.source().query(QueryBuilders.matchAllQuery());
        //聚合条件
        searchRequest.source().aggregation(AggregationBuilders.terms("品牌聚合").field("brand").size(10));
        //发送搜索请求
        SearchResponse response = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);

        //处理结果
        //获取品牌聚合结果
        Aggregation aggregation = response.getAggregations().get("品牌聚合");
        //类型强转，获取更多功能
        Terms terms = (Terms) aggregation;
        //获取桶
        List<? extends Terms.Bucket> buckets = terms.getBuckets();
        //循环桶，获取每个分组结果
        for (Terms.Bucket bucket : buckets) {
    
            String key = bucket.getKeyAsString();
            long count = bucket.getDocCount();
            System.out.println("品牌名:" + key + ",数量为" + count);
        }
    }

    @Test
    public void testAggs() throws IOException {
    
        //创建搜索请求对象
        SearchRequest searchRequest = new SearchRequest("hotel");
        //查询条件
        searchRequest.source().query(QueryBuilders.matchAllQuery());
        //聚合条件
        searchRequest.source().aggregation(AggregationBuilders.terms("品牌聚合").field("brand").size(10). //按照品牌字段进行聚合,显示10条
                subAggregation(AggregationBuilders.stats("stats").field("price"))); //子聚合 ,按照价格做聚合函数 min max avg count sum

        //发送查询请求
        SearchResponse response = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);

        //处理结果
        //获取品牌聚合结果
        Aggregation aggregation = response.getAggregations().get("品牌聚合");
        //类型强转，获取更多功能
        Terms terms = (Terms) aggregation;
        //获取桶
        List<? extends Terms.Bucket> buckets = terms.getBuckets();
        //循环桶，获取每个分组结果
        for (Terms.Bucket bucket : buckets) {
    
            String key = bucket.getKeyAsString();
            long count = bucket.getDocCount();
            System.out.println("品牌名:" + key + ",数量为" + count);
            //子聚合结果
            Stats stats = bucket.getAggregations().get("stats");
            System.out.println("价格平均值:" + stats.getAvgAsString());
            System.out.println("价格总和:" + stats.getSumAsString());
            System.out.println("价格最大值:" + stats.getMaxAsString());
            System.out.println("价格最小值:" + stats.getMinAsString());
        }
    }

2.数据同步

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

2.1.思路分析

常见的数据同步方案有三种：

同步调用
异步通知
监听binlog

2.1.1.同步调用

方案一：同步调用

在这里插入图片描述
基本步骤如下：

hotel-demo对外提供接口，用来修改elasticsearch中的数据
酒店管理服务在完成数据库操作后，直接调用hotel-demo提供的接口，

2.1.2.异步通知

方案二：异步通知

在这里插入图片描述

流程如下：

hotel-admin对mysql数据库数据完成增、删、改后，发送MQ消息
hotel-demo监听MQ，接收到消息后完成elasticsearch数据修改

2.1.3.监听binlog

方案三：监听binlog

在这里插入图片描述

流程如下：

给mysql开启binlog功能
mysql完成增、删、改操作都会记录在binlog中
hotel-demo基于canal监听binlog变化，实时更新elasticsearch中的内容

2.1.4.选择

方式一：同步调用

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

方式二：异步通知

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

方式三：监听binlog

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

3.集群

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

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

ES集群相关概念:

集群（cluster）：一组拥有共同的 cluster name 的节点。
节点（node) ：集群中的一个 Elasticearch 实例
分片（shard）：索引可以被拆分为不同的部分进行存储，称为分片。在集群环境下，一个索引的不同分片可以拆分到不同的节点中
解决问题：数据量太大，单点存储量有限的问题。
主分片（Primary shard）：相对于副本分片的定义。
副本分片（Replica shard）每个主分片可以有一个或者多个副本，数据和主分片一样。

数据备份可以保证高可用，但是每个分片备份一份，所需要的节点数量就会翻一倍，成本实在是太高了！

为了在高可用和成本间寻求平衡，我们可以这样做：