elasticsearch多字段聚合实现方式

末蓝、 2024-04-01 18:11 84阅读 0赞

#### 文章目录 ####

*  1、背景
 *  2、实现多字段聚合的思路
 *  3、需求
 *  4、数据准备
 *   *  4.1 创建索引
     *  4.2 准备数据
 *  5、实现方式
 *   *  5.1 multi\_terms实现
     *   *  5.1.1 dsl
         *  5.1.2 java 代码
         *  5.1.3 运行结果
     *  5.2 script实现
     *   *  5.2.1 dsl
         *  5.2.2 java代码
         *  5.2.3 运行结果
     *  5.3 通过copyto实现
     *  5.5 通过pipeline来实现
     *   *  5.4.1 创建mapping
         *  5.4.2 创建pipeline
         *  5.4.3 插入数据
         *  5.4.4 聚合dsl
         *  5.4.5 运行结果
 *  6、实现代码
 *  7、参考文档

## 1、背景 ##

我们知道在`sql`中是可以实现 `group by 字段a,字段b`，那么这种效果在`elasticsearch`中该如何实现呢？此处我们记录在`elasticsearch`中的3种方式来实现这个效果。

## 2、实现多字段聚合的思路 ##

![实现多字段聚合的思路][1014da1aeb2e4a68bf7170f7f4687e9e.png]  
图片来源：[https://www.elastic.co/guide/en/elasticsearch/reference/current/search-aggregations-bucket-terms-aggregation.html][https_www.elastic.co_guide_en_elasticsearch_reference_current_search-aggregations-bucket-terms-aggregation.html]  
从上图中，我们可以知道，可以通过3种方式来实现 多字段的聚合操作。

## 3、需求 ##

根据省(`province`)和性别(`sex`)来进行聚合，然后根据聚合后的每个桶的数据，在根据每个桶中的最大年龄(`age`)来进行倒序排序。

## 4、数据准备 ##

### 4.1 创建索引 ###

PUT /index_person
    {
      "settings": {
        "number_of_shards": 1
      },
      "mappings": {
        "properties": {
          "id": {
            "type": "long"
          },
          "name": {
            "type": "keyword"
          },
          "province": {
            "type": "keyword"
          },
          "sex": {
            "type": "keyword"
          },
          "age": {
            "type": "integer"
          },
          "address": {
            "type": "text",
            "analyzer": "ik_max_word",
            "fields": {
              "keyword": {
                "type": "keyword",
                "ignore_above": 256
              }
            }
          }
        }
      }
    }

### 4.2 准备数据 ###

PUT /_bulk
    {"create":{"_index":"index_person","_id":1}}
    {"id":1,"name":"张三","sex":"男","age":20,"province":"湖北","address":"湖北省黄冈市罗田县匡河镇"}
    {"create":{"_index":"index_person","_id":2}}
    {"id":2,"name":"李四","sex":"男","age":19,"province":"江苏","address":"江苏省南京市"}
    {"create":{"_index":"index_person","_id":3}}
    {"id":3,"name":"王武","sex":"女","age":25,"province":"湖北","address":"湖北省武汉市江汉区"}
    {"create":{"_index":"index_person","_id":4}}
    {"id":4,"name":"赵六","sex":"女","age":30,"province":"北京","address":"北京市东城区"}
    {"create":{"_index":"index_person","_id":5}}
    {"id":5,"name":"钱七","sex":"女","age":16,"province":"北京","address":"北京市西城区"}
    {"create":{"_index":"index_person","_id":6}}
    {"id":6,"name":"王八","sex":"女","age":45,"province":"北京","address":"北京市朝阳区"}

## 5、实现方式 ##

### 5.1 multi\_terms实现 ###

#### 5.1.1 dsl ####

GET /index_person/_search
    {
      "size": 0,
      "aggs": {
        "agg_province_sex": {
          "multi_terms": {
            "size": 10,
            "shard_size": 25,
            "order":{
              "max_age": "desc"    
            },
            "terms": [
              {
                "field": "province",
                "missing": "defaultProvince"
              },
              {
                "field": "sex"
              }
            ]
          },
          "aggs": {
            "max_age": {
              "max": {
                "field": "age"
              }
            }
          }
        }
      }
    }

#### 5.1.2 java 代码 ####

@Test
        @DisplayName("多term聚合-根据省和性别聚合，然后根据最大年龄倒序")
        public void agg01() throws IOException {
        
    
            SearchRequest searchRequest = new SearchRequest.Builder()
                    .size(0)
                    .index("index_person")
                    .aggregations("agg_province_sex", agg ->
                            agg.multiTerms(multiTerms ->
                                            multiTerms.terms(term -> term.field("province"))
                                                    .terms(term -> term.field("sex"))
                                                    .order(new NamedValue<>("max_age", SortOrder.Desc))
                                    )
                                    .aggregations("max_age", ageAgg ->
                                            ageAgg.max(max -> max.field("age")))
    
                    )
                    .build();
            System.out.println(searchRequest);
            SearchResponse<Object> response = client.search(searchRequest, Object.class);
            System.out.println(response);
        }

#### 5.1.3 运行结果 ####

![运行结果][431215e24af54e0e9ebdf6e3d5cc6690.png]

### 5.2 script实现 ###

#### 5.2.1 dsl ####

GET /index_person/_search
    {
      "size": 0,
      "runtime_mappings": {
        "runtime_province_sex": {
          "type": "keyword",
          "script": """
              String province = doc['province'].value;
              String sex = doc['sex'].value;
              emit(province + '|' + sex);
          """
        }
      },
      "aggs": {
        "agg_province_sex": {
          "terms": {
            "field": "runtime_province_sex",
            "size": 10,
            "shard_size": 25,
            "order": {
              "max_age": "desc"
            }
          },
          "aggs": {
            "max_age": {
              "max": {
                "field": "age"
              }
            }
          }
        }
      }
    }

#### 5.2.2 java代码 ####

@Test
        @DisplayName("多term聚合-根据省和性别聚合，然后根据最大年龄倒序")
        public void agg02() throws IOException {
        
    
            SearchRequest searchRequest = new SearchRequest.Builder()
                    .size(0)
                    .index("index_person")
                    .runtimeMappings("runtime_province_sex", field -> {
        
                        field.type(RuntimeFieldType.Keyword);
                        field.script(script -> script.inline(new InlineScript.Builder()
                                .lang(ScriptLanguage.Painless)
                                .source("String province = doc['province'].value;\n" +
                                        "          String sex = doc['sex'].value;\n" +
                                        "          emit(province + '|' + sex);")
                                .build()));
                        return field;
                    })
                    .aggregations("agg_province_sex", agg ->
                            agg.terms(terms ->
                                            terms.field("runtime_province_sex")
                                                    .size(10)
                                                    .shardSize(25)
                                                    .order(new NamedValue<>("max_age", SortOrder.Desc))
                                    )
                                    .aggregations("max_age", minAgg ->
                                            minAgg.max(max -> max.field("age")))
                    )
                    .build();
            System.out.println(searchRequest);
            SearchResponse<Object> response = client.search(searchRequest, Object.class);
            System.out.println(response);
        }

#### 5.2.3 运行结果 ####

![运行结果][2e90c46ff82f43b2be82de4cb83a806e.png]

### 5.3 通过copyto实现 ###

`我本地测试过，通过copyto没实现，此处故先不考虑`

### 5.5 通过pipeline来实现 ###

`实现思路：`  
创建mapping时，多创建一个字段`pipeline_province_sex`，该字段的值由创建数据时指定`pipeline`来生产。

#### 5.4.1 创建mapping ####

PUT /index_person
    {
      "settings": {
        "number_of_shards": 1
      },
      "mappings": {
        "properties": {
          "id": {
            "type": "long"
          },
          "name": {
            "type": "keyword"
          },
          "province": {
            "type": "keyword"
          },
          "sex": {
            "type": "keyword"
          },
          "age": {
            "type": "integer"
          },
          "pipeline_province_sex":{
            "type": "keyword"
          },
          "address": {
            "type": "text",
            "analyzer": "ik_max_word",
            "fields": {
              "keyword": {
                "type": "keyword",
                "ignore_above": 256
              }
            }
          }
        }
      }
    }

此处指定了一个字段`pipeline_province_sex`，该字段的值会由`pipeline`来处理。

#### 5.4.2 创建pipeline ####

PUT _ingest/pipeline/pipeline_index_person_provice_sex
    {
      "description": "将provice和sex的值拼接起来",
      "processors": [
        {
          "set": {
            "field": "pipeline_province_sex",
            "value": ["{
       {province}}", "{
       {sex}}"]
          }, 
          "join": {
            "field": "pipeline_province_sex",
            "separator": "|"
          }
        }
      ]
    }

#### 5.4.3 插入数据 ####

PUT /_bulk?pipeline=pipeline_index_person_provice_sex
    {"create":{"_index":"index_person","_id":1}}
    {"id":1,"name":"张三","sex":"男","age":20,"province":"湖北","address":"湖北省黄冈市罗田县匡河镇"}
    {"create":{"_index":"index_person","_id":2}}
    {"id":2,"name":"李四","sex":"男","age":19,"province":"江苏","address":"江苏省南京市"}
    {"create":{"_index":"index_person","_id":3}}
    {"id":3,"name":"王武","sex":"女","age":25,"province":"湖北","address":"湖北省武汉市江汉区"}
    {"create":{"_index":"index_person","_id":4}}
    {"id":4,"name":"赵六","sex":"女","age":30,"province":"北京","address":"北京市东城区"}
    {"create":{"_index":"index_person","_id":5}}
    {"id":5,"name":"钱七","sex":"女","age":16,"province":"北京","address":"北京市西城区"}
    {"create":{"_index":"index_person","_id":6}}
    {"id":6,"name":"王八","sex":"女","age":45,"province":"北京","address":"北京市朝阳区"}

**注意：** 此处的插入需要指定上一步的`pipeline`  
`PUT /_bulk?pipeline=pipeline_index_person_provice_sex`

#### 5.4.4 聚合dsl ####

GET /index_person/_search
    {
      "size": 0,
      "aggs": {
        "agg_province_sex": {
          "terms": {
            "field": "pipeline_province_sex",
            "size": 10,
            "shard_size": 25,
            "order": {
              "max_age": "desc"   
            }
          }, 
          "aggs": {
            "max_age": {
              "max": {
                "field": "age"
              }
            }
          }
        }
      }
    }

#### 5.4.5 运行结果 ####

![运行结果][b89ca3a8e47d4155b306765419fe0fc5.png]

## 6、实现代码 ##

[https://gitee.com/huan1993/spring-cloud-parent/blob/master/es/es8-api/src/main/java/com/huan/es8/aggregations/bucket/MultiTermsAggs.java][https_gitee.com_huan1993_spring-cloud-parent_blob_master_es_es8-api_src_main_java_com_huan_es8_aggregations_bucket_MultiTermsAggs.java]

## 7、参考文档 ##

1.  [https://www.elastic.co/guide/en/elasticsearch/reference/current/search-aggregations-bucket-terms-aggregation.html][https_www.elastic.co_guide_en_elasticsearch_reference_current_search-aggregations-bucket-terms-aggregation.html]

[1014da1aeb2e4a68bf7170f7f4687e9e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/8f113955f3524511bdc19640a833e0d6.png
[https_www.elastic.co_guide_en_elasticsearch_reference_current_search-aggregations-bucket-terms-aggregation.html]: https://www.elastic.co/guide/en/elasticsearch/reference/current/search-aggregations-bucket-terms-aggregation.html
[431215e24af54e0e9ebdf6e3d5cc6690.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/60254b40c0f243fe80a2343a75301c99.png
[2e90c46ff82f43b2be82de4cb83a806e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/3a320602ba3f405a9de5e14abfad412f.png
[b89ca3a8e47d4155b306765419fe0fc5.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/a86598ef247f4195813c4d46f2daa689.png
[https_gitee.com_huan1993_spring-cloud-parent_blob_master_es_es8-api_src_main_java_com_huan_es8_aggregations_bucket_MultiTermsAggs.java]: https://gitee.com/huan1993/spring-cloud-parent/blob/master/es/es8-api/src/main/java/com/huan/es8/aggregations/bucket/MultiTermsAggs.java