ElasticSearch聚合操作详解

聚合操作

聚合查询的语法结构与其他查询相似，通常包含以下部分：

• 查询条件：指定需要聚合的文档，可以使用标准的 Elasticsearch 查询语法，如 term、match、range 等等。
• 聚合函数：指定要执行的聚合操作，如 sum、avg、min、max、terms、date_histogram 等等。每个聚合命令都会生成一个聚合结果。
• 聚合嵌套：聚合命令可以嵌套，以便更细粒度地分析数据。

GET <index_name>/_search
{"aggs": {"<aggs_name>": { #  聚合名称需要自己定义"<agg_type>": {		# 聚合种类，比如是桶聚合（terms）或者是指标聚合（avg、sum、min、max等）"field": "<field_name>"			#  field_name 字段名称或者叫域名。}}}
}

聚合操作一般都是操作的术语级别的字段，一般不会操作text类型的字段

聚合的分类

• Metric Aggregation：—些数学运算，可以对文档字段进行统计分析，类比Mysql中的 min(), max(), sum() 操作。

GET /sys_user/_search
{"aggs": {"hs_avg_age": {"avg": {		# 这里还可以使用 max   min 等等"field": "age"}}}
}

• Bucket Aggregation： 一些满足特定条件的文档的集合放置到一个桶里，每一个桶关联一个key，类比Mysql中的group by操作。

  # 使用terms关键字
  # 类似于关系型数据库的分组操作
  # 先分组 再求和
  GET /sys_user/_search
  {"aggs": {"hs_terms_age": {"terms": {  # 使用terms关键字"field": "age"}}}
  }
  

  

• Pipeline Aggregation：对其他的聚合结果进行二次聚合

测试数据

聚合操作一般都是操作术语级别的字段，比较少操作text类型的字段

DELETE /employees
#创建索引库
PUT /employees
{"mappings": {"properties": {"age":{"type": "integer"},"gender":{"type": "keyword"},"job":{"type" : "text","fields" : {"keyword" : {"type" : "keyword","ignore_above" : 50}}},"name":{"type": "keyword"},"salary":{"type": "integer"}}}
}PUT /employees/_bulk
{ "index" : {  "_id" : "1" } }
{ "name" : "Emma","age":32,"job":"Product Manager","gender":"female","salary":35000 }
{ "index" : {  "_id" : "2" } }
{ "name" : "Underwood","age":41,"job":"Dev Manager","gender":"male","salary": 50000}
{ "index" : {  "_id" : "3" } }
{ "name" : "Tran","age":25,"job":"Web Designer","gender":"male","salary":18000 }
{ "index" : {  "_id" : "4" } }
{ "name" : "Rivera","age":26,"job":"Web Designer","gender":"female","salary": 22000}
{ "index" : {  "_id" : "5" } }
{ "name" : "Rose","age":25,"job":"QA","gender":"female","salary":18000 }
{ "index" : {  "_id" : "6" } }
{ "name" : "Lucy","age":31,"job":"QA","gender":"female","salary": 25000}
{ "index" : {  "_id" : "7" } }
{ "name" : "Byrd","age":27,"job":"QA","gender":"male","salary":20000 }
{ "index" : {  "_id" : "8" } }
{ "name" : "Foster","age":27,"job":"Java Programmer","gender":"male","salary": 20000}
{ "index" : {  "_id" : "9" } }
{ "name" : "Gregory","age":32,"job":"Java Programmer","gender":"male","salary":22000 }
{ "index" : {  "_id" : "10" } }
{ "name" : "Bryant","age":20,"job":"Java Programmer","gender":"male","salary": 9000}
{ "index" : {  "_id" : "11" } }
{ "name" : "Jenny","age":36,"job":"Java Programmer","gender":"female","salary":38000 }
{ "index" : {  "_id" : "12" } }
{ "name" : "Mcdonald","age":31,"job":"Java Programmer","gender":"male","salary": 32000}
{ "index" : {  "_id" : "13" } }
{ "name" : "Jonthna","age":30,"job":"Java Programmer","gender":"female","salary":30000 }
{ "index" : {  "_id" : "14" } }
{ "name" : "Marshall","age":32,"job":"Javascript Programmer","gender":"male","salary": 25000}
{ "index" : {  "_id" : "15" } }
{ "name" : "King","age":33,"job":"Java Programmer","gender":"male","salary":28000 }
{ "index" : {  "_id" : "16" } }
{ "name" : "Mccarthy","age":21,"job":"Javascript Programmer","gender":"male","salary": 16000}
{ "index" : {  "_id" : "17" } }
{ "name" : "Goodwin","age":25,"job":"Javascript Programmer","gender":"male","salary": 16000}
{ "index" : {  "_id" : "18" } }
{ "name" : "Catherine","age":29,"job":"Javascript Programmer","gender":"female","salary": 20000}
{ "index" : {  "_id" : "19" } }
{ "name" : "Boone","age":30,"job":"DBA","gender":"male","salary": 30000}
{ "index" : {  "_id" : "20" } }
{ "name" : "Kathy","age":29,"job":"DBA","gender":"female","salary": 20000}

Metric Aggregation

—些数学运算，可以对文档字段进行统计分析，类比Mysql中的 min(), max(), sum() 操作。

• 单值分析︰只输出一个分析结果

  • min, max, avg, sum
  • Cardinality（类似distinct Count)

• 多值分析:输出多个分析结果

  • stats（统计）, extended stats

  • percentile （百分位）, percentile rank

  • top hits(排在前面的示例)

如果只用到聚合信息，而不需要源json数据，那么就可以通过"size": 0 来指定不输出源json数据

查询员工的最低、最高和平均工资

GET /employees/_search
{"size": 0, "aggs": {"hs_min_salary": {"min": {"field": "salary"}},"hs_max_salary": {"max": {"field": "salary"}},"hs_avg_salary": {"avg": {"field": "salary"}}}
}

对salary进行统计

# 对salary进行统计
GET /employees/_search
{"size": 0,"aggs": {"hs_stats_salary": {"stats": {			# 使用stats关键字"field": "salary"}}}
}

cardinate对搜索结果去重

# cardinate对搜索结果去重
# 需要使用keyword子类型，不能直接使用text类型字段
POST /employees/_search
{"size": 0,"aggs": {"hs_cardinate": {"cardinality": {"field": "job.keyword"}}}
}

Bucket Aggregation

一些满足特定条件的文档的集合放置到一个桶里，每一个桶关联一个key，类比Mysql中的group by操作。

按照一定的规则，将文档分配到不同的桶中，从而达到分类的目的。

• Terms，需要字段支持filedata

  • keyword 默认支持fielddata
  • text需要在Mapping 中开启fielddata，会按照分词后的结果进行分桶

• 数字类型

  • Range / Data Range
  • Histogram（直方图） / Date Histogram

• 支持嵌套: 也就在桶里再做分桶

获取job的分类信息

其实就是对某个字段分组，然后就各个组下的文档总数

# 获取job字段的分类信息
# text类型的字段需要再mappings中开启fielddata，这里使用job.keyword
GET /employees/_search
{"size": 0,"aggs": {"hs_job": {"terms": {"field": "job.keyword"}}}
}

聚合可配置属性有：

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

默认情况下，Bucket聚合会统计Bucket内的文档数量，记为_count，并且按照count降序排序。我们可以指定order属性，自定义聚合的排序方式：

GET /employees/_search
{"size": 0,"aggs": {"hs_job": {"terms": {"field": "job.keyword","size": 5,				# 只显示5条聚合结果"order": {"_count": "asc"		# 按升序排序}}}}
}

限定聚合范围

其实就是先查询过滤，然后在聚合

# 只对salary在10000元以上的文档聚合
GET /employees/_search
{"size": 0,"query": {"range": {"salary": {"gte": 10000}}}, "aggs": {"hs_job": {"terms": {"field": "job.keyword","size": 10,"order": {"_count": "desc"}}}}
}

注意：对 Text 字段进行 terms 聚合查询，会失败抛出异常

GET /employees/_search
{"size": 0,"aggs": {"hs_job": {"terms": {"field": "job"}}}
}

对 Text 字段打开 fielddata，支持terms aggregation

# 先查询索引mappings中的数据
GET /employees/_mapping# 再修改对应字段
PUT /employees/_mapping
{"properties": {"job": {"type": "text","fielddata": true,		# 添加fielddata为true"fields": {"keyword": {"type": "keyword","ignore_above": 50}}}}
}

对job.keyword 和 job 进行 terms 聚合，分桶的总数并不一样，如下图所示

Range & Histogram聚合

• 按照数字的范围，进行分桶
• 在Range Aggregation中，可以自定义Key

Range 示例：按照工资的 Range 分桶

# range分桶，可以自己定义 key
POST employees/_search
{"size": 0,"aggs": {"hs_salary_range": {"range": {			# 使用range关键字"field":"salary","ranges":[{"to":10000,"key": "hs_key 0~10000"		# 自定义key},{"from":10000,"to":20000},{"from":20000}]}}}
}

Histogram示例：按照工资的间隔分桶

# 工资0到10万，以 5000一个区间进行分桶
POST employees/_search
{"size": 0,"aggs": {"hs_salary_histogram": {"histogram": {			# 使用histogram关键字"field": "salary","interval": 5000,		# 指定区间"extended_bounds": {	# 指定总范围"min": 0,"max": 100000}}}}
}

聚合嵌套

top_hits应用场景: 当获取分桶后，桶内最匹配的顶部文档列表

# 查询不同工种中，年纪最大的3个员工的具体信息
POST employees/_search
{"size": 0,"aggs": {"hs_jobs": {"terms": {"field": "job.keyword"},"aggs": {		# 和terms平级，"hs_old_employee": {"top_hits": {		# 使用top_hits关键字"size": 3,"sort": [{"age": {"order": "desc"}}]}}}}}
}

# 按照工作类型分桶，并统计工资信息
GET /employees/_search
{"size": 0,"aggs": {"hs_job": {"terms": {"field": "job.keyword"},"aggs": {"hs_salary": {"stats": {"field": "salary"}}}}}
}

# 多次嵌套。根据工作类型分桶，然后按照性别分桶，计算工资的统计信息
GET /employees/_search
{"size": 0,"aggs": {"hs_job": {"terms": {"field": "job.keyword","size": 10},"aggs": {"hs_gender": {"terms": {"field": "gender","size": 10},"aggs": {"hs_salary": {"stats": {"field": "salary"}}}}}}}
}

Pipeline Aggregation

支持对聚合分析的结果，再次进行聚合分析。

Pipeline 的分析结果会输出到原结果中，根据位置的不同，分为两类：

• Sibling - 结果和现有分析结果同级

  • Max，min，Avg ， Sum Bucket
  • Stats，Extended Status Bucket
  • Percentiles Bucket

• Parent -结果内嵌到现有的聚合分析结果之中

  • Derivative(求导)
  • Cumultive Sum(累计求和)
  • Moving Function(移动平均值 )

min_bucket示例

# 平均工资最低的工种
GET /employees/_search
{"size": 0,"aggs": {"hs_job": {"terms": {"field": "job.keyword"},"aggs": {"hs_avg_salary": {"avg": {"field": "salary"}}}},"hs_min_salary_by_job": {		# 这个要和上面的hs_job平级"min_bucket": {		#  使用 min_bucket 关键字  求之前结果的最小值"buckets_path": "hs_job>hs_avg_salary"	 #  通过bucket_path关键字指定路径}}}
}

Stats示例

# 平均工资的统计分析 , 使用 stats_bucket 关键字
GET /employees/_search
{"size": 0,"aggs": {"hs_job": {"terms": {"field": "job.keyword"},"aggs": {"hs_avg_salary": {"avg": {"field": "salary"}}}},"hs_min_salary_by_job": {"stats_bucket": {		# 使用stats_bucket关键字"buckets_path": "hs_job>hs_avg_salary"}}}
}

percentiles示例

使用 percentiles_bucket 关键字

# 平均工资的百分位数
GET /employees/_search
{"size": 0,"aggs": {"hs_job": {"terms": {"field": "job.keyword"},"aggs": {"hs_avg_salary": {"avg": {"field": "salary"}}}},"hs_min_salary_by_job": {"percentiles_bucket": {  # 使用 percentiles_bucket 关键字"buckets_path": "hs_job>hs_avg_salary"}}}
}

Cumulative_sum示例

Cumulative_sum 累计求和

GET /employees/_search
{"size": 0,"aggs": {"hs_age": {"histogram": {  			# histogram直方图"field": "age","min_doc_count": 0, "interval": 1			# age年龄字段 每次自增1},"aggs": {"hs_avg_salary": {"avg": {"field": "salary"}},"hs_cumulative_salary": {"cumulative_sum": {"buckets_path": "hs_avg_salary"}}}}}
}

聚合的作用范围

ES聚合分析的默认作用范围是query的查询结果集，同时ES还支持以下方式改变聚合的作用范围：

• Filter：在聚合前进一步查询过滤条件
• Post Filter ：能找到聚合后，符合指定条件的文档数据
• Global：即使之前通过查询过滤掉了一些数据，global还是能对所有数据进行聚合操作

#Query
# 先查询 在对满足查询条件的文档进行聚合操作
POST employees/_search
{"size": 0,"query": {"range": {"age": {"gte": 26}}},"aggs": {"hs_jobs": {"terms": {"field": "job.keyword"}}}
}#Filter
# 功能和上方类似，只不过是把查询过滤条件放在了filter中，建议还是使用上方query的方式
POST employees/_search
{"size": 0,"aggs": {"hs_older_person": {			# 对年龄>=26的员工进行聚合操作"filter": {"range": {"age": {"from": 26}}},"aggs": {"jobs": {"terms": {"field": "job.keyword"}}}},"hs_all_jobs": {		# 对所有员工进行聚合操作"terms": {"field": "job.keyword"}}}
}

post_filter. 一条语句，找出所有的job类型。还能找到聚合后符合条件的结果

# 找到聚合后复合条件的文档，显示出来。 这里就是找job为Javascript Programmer的员工信息
POST employees/_search
{"aggs": {"hs_jobs": {"terms": {"field": "job.keyword"}}},"post_filter": {"match": {"job.keyword": "Javascript Programmer"}}
}

global示例

# 大于等于33岁的只有3个员工
# 但是global中还是算的所有员工的平均工资
POST employees/_search
{"size": 0,"query": {"range": {"age": {"gte": 33}}},"aggs": {"hs_jobs": {"terms": {"field":"job.keyword"}},"hs_all":{"global":{},"aggs":{"hs_salary_avg":{"avg":{"field":"salary"}}}}}
}

排序

指定order，按照count和key进行排序：

• 默认情况，按照count降序排序
• 指定size，就能返回相应的桶

#排序 order
#count and key
POST employees/_search
{"size": 0,"query": {		# 查询 过滤不满足的文档"range": {"age": {"gte": 20}}},"aggs": {"hs_jobs": {"terms": {"field":"job.keyword",		# 对job进行分组"order":[{"_count":"asc"},			# 先按count升序排序，再按key降序排序{"_key":"desc"}]}}}
}

# 也可以对嵌套聚合的值进行排序
# 比如对先按job分组，再求出每组的平均工资，在按平均工资降序排序
POST employees/_search
{"size": 0,"aggs": {"hs_jobs": {"terms": {"field": "job.keyword","order": [{"hs_avg_salary": "desc"}]     # 等同于{"hs_avg_salary.value": "desc"}},"aggs": {"hs_avg_salary": {"avg": {"field": "salary"}}}}}
}

# 按job分组，再统计每组下的工作。最后按统计结果中的最小值排序
POST employees/_search
{"size": 0,"aggs": {"jobs": {"terms": {"field": "job.keyword","order": [{"hs_stats_salary.min": "desc"}]},"aggs": {"hs_stats_salary": {"stats": {"field": "salary"}}}}}
}

ES聚合分析不精准原因分析

ElasticSearch在对海量数据进行聚合分析的时候会损失搜索的精准度来满足实时性的需求。

Terms聚合分析的执行流程：

各个shard分片中先对结果进行处理，然后把数据传输给协调者节点，协调节点再对数据进行汇总归并，再响应请求

不精准的原因： 数据分散到多个分片，聚合是每个分片的取 Top X，导致结果不精准。

下方案例中，求和返回数据量最大的3个数据，D的数据在两个shard上，结果是6，但是返回的数据中没有D

思考：如何提高聚合精确度？

方案1：设置主分片为1

注意7.x版本已经默认为1。

适用场景：数据量小的小集群规模业务场景。

对于数据量大的情况，肯定不可能只使用一个shard

方案2：调大 shard_size 值

比如客户一次请求要查询最大的三条记录，但是我每个分片可以给协调节点单分片排序后的 5条数据

设置 shard_size 为比较大的值，官方推荐：size*1.5+10。shard_size 值越大，结果越趋近于精准聚合结果值。此外，还可以通过show_term_doc_count_error参数显示最差情况下的错误值，用于辅助确定 shard_size 大小。

• size：是聚合结果的返回值，客户期望返回聚合排名前三，size值就是 3。
• shard_size: 每个分片上聚合的数据条数。shard_size 原则上要大于等于 size

适用场景：数据量大、分片数多的集群业务场景。

测试，使用kibana的测试数据

使用reindex重建索引，因为kibana提供的测试数据默认是一个shard，我重建索引，指定多个分片

GET kibana_sample_data_flights
PUT /my_flights
{"settings": {"number_of_shards": 20}, "mappings": {"properties" : {"AvgTicketPrice" : {"type" : "float"},"Cancelled" : {"type" : "boolean"},"Carrier" : {"type" : "keyword"},"Dest" : {"type" : "keyword"},"DestAirportID" : {"type" : "keyword"},"DestCityName" : {"type" : "keyword"},"DestCountry" : {"type" : "keyword"},"DestLocation" : {"type" : "geo_point"},"DestRegion" : {"type" : "keyword"},"DestWeather" : {"type" : "keyword"},"DistanceKilometers" : {"type" : "float"},"DistanceMiles" : {"type" : "float"},"FlightDelay" : {"type" : "boolean"},"FlightDelayMin" : {"type" : "integer"},"FlightDelayType" : {"type" : "keyword"},"FlightNum" : {"type" : "keyword"},"FlightTimeHour" : {"type" : "keyword"},"FlightTimeMin" : {"type" : "float"},"Origin" : {"type" : "keyword"},"OriginAirportID" : {"type" : "keyword"},"OriginCityName" : {"type" : "keyword"},"OriginCountry" : {"type" : "keyword"},"OriginLocation" : {"type" : "geo_point"},"OriginRegion" : {"type" : "keyword"},"OriginWeather" : {"type" : "keyword"},"dayOfWeek" : {"type" : "integer"},"timestamp" : {"type" : "date"}}}
}# 使用reindex进行数据迁移
POST _reindex
{"source": {"index": "kibana_sample_data_flights"},"dest": {"index": "my_flights"}
}# 之前的索引就可以删除了
DELETE /kibana_sample_data_flights

GET my_flights/_search
{"size": 0,"aggs": {"hs_weather": {"terms": {"field": "OriginWeather","size": 5,"show_term_doc_count_error": true		# 开启show_term_doc_count_error参数}}}
}GET my_flights/_search
{"size": 0,"aggs": {"hs_weather": {"terms": {"field": "OriginWeather","size": 5,"shard_size": 10,			# 使用shard_size参数"show_term_doc_count_error": true  # 开启show_term_doc_count_error参数}}}
}

在Terms Aggregation的返回中有两个特殊的数值：

• doc_count_error_upper_bound : 被遗漏的term 分桶，包含的文档，有可能的最大值

• sum_other_doc_count: 除了返回结果 bucket的terms以外，其他 terms 的文档总数（总数-返回的总数)

方案3：将size设置为全量值，来解决精度问题

将size设置为2的32次方减去1也就是分片支持的最大值，来解决精度问题。

原因：1.x版本，size等于 0 代表全部，高版本取消 0 值，所以设置了最大值（大于业务的全量值）。

全量带来的弊端就是：如果分片数据量极大，这样做会耗费巨大的CPU 资源来排序，而且可能会阻塞网络。

适用场景：对聚合精准度要求极高的业务场景，由于性能问题，不推荐使用。

方案4：使用Clickhouse/ Spark 进行精准聚合

适用场景：数据量非常大、聚合精度要求高、响应速度快的业务场景。

聚合性能优化

启用 eager global ordinals 提升高基数聚合性能

适用场景：高基数聚合 。

高基数聚合场景中的高基数含义：一个字段包含很大比例的唯一值。

适合启动eager_global_ordinals的场景有：

• 基于 keyword，ip 等字段的分桶聚合
• 基于text 字段的分桶聚合（前提条件是：fielddata 开启）。
• 基于父子文档 Join 类型的 has_child 查询和 父聚合。

global ordinals 使用一个数值代表字段中的字符串值，然后为每一个数值分配一个 bucket（分桶）。

global ordinals 的本质是：

启用 eager_global_ordinals 时，会在刷新分片时构建全局序号。这将构建全局序号的成本从搜索阶段转移到了数据索引化（写入）阶段。

创建索引的同时开启：eager_global_ordinals。

PUT /my-index
{"mappings": {"properties": {"tags": {"type": "keyword","eager_global_ordinals": true 	# 开启eager_global_ordinals}}}
}

开启 eager_global_ordinals 会影响写入性能，因为每次刷新时都会创建新的全局序号。为了最大程度地减少由于频繁刷新建立全局序号而导致的额外开销，请调大刷新间隔 refresh_interval。

动态调整刷新频率的方法如下：

PUT my-index/_settings
{"index": {"refresh_interval": "30s"}
}

本质是：以空间换时间。

插入数据时对索引进行预排序

• Index sorting （索引排序）可用于在插入时对索引进行预排序，而不是在查询时再对索引进行排序，这将提高范围查询（range query）和排序操作的性能。
• 在 Elasticsearch 中创建新索引时，可以配置如何对每个分片内的段进行排序。
• 这是 Elasticsearch 6.X 之后版本才有的特性。

PUT /my_index
{"settings": {"index":{"sort.field": "create_time",  # 指定按照哪个字段进行排序"sort.order": "desc"		# 降序}},"mappings": {"properties": {"create_time":{		# 创建字段，字段名和上方对应"type": "date"}}}
}

注意：预排序将增加 Elasticsearch 写入的成本。

在某些用户特定场景下，开启索引预排序会导致大约 40%-50% 的写性能下降。也就是说，如果用户场景更关注写性能的业务，开启索引预排序不是一个很好的选择。

使用节点查询缓存

节点查询缓存（Node query cache）可用于有效缓存过滤器（filter）操作的结果。

如果多次执行同一 filter 操作，这将很有效，但是即便更改过滤器中的某一个值，也将意味着需要计算新的过滤器结果。

例如，由于 “now” 值一直在变化，因此无法缓存在过滤器上下文中使用 “now” 的查询。

那怎么使用缓存呢？通过在 now 字段上应用 datemath 格式将其四舍五入到最接近的分钟/小时等，可以使此类请求更具可缓存性，以便可以对筛选结果进行缓存。

PUT /my_index/_doc/1
{"create_time":"2022-05-11T16:30:55.328Z"
}#下面的示例无法使用缓存
GET /my_index/_search
{"query":{"constant_score": {"filter": {"range": {"create_time": {"gte": "now-1h","lte": "now"}}}}}
}# 下面的示例就可以使用节点查询缓存。
GET /my_index/_search
{"query":{"constant_score": {"filter": {"range": {"create_time": {"gte": "now-1h/m",     # 在最后增加了/m    m代表分钟"lte": "now/m"}}}}}
}

上述示例中的now-1h/m 就是 datemath 的格式。

如果当前时间 now 是：16:31:29，那么range query 将匹配 my_date 介于：15:31:00 和 15:31:59 之间的时间数据。同理，聚合的前半部分 query 中如果有基于时间查询，或者后半部分 aggs 部分中有基于时间聚合的，建议都使用 datemath 方式做缓存处理以优化性能。

使用分片请求缓存

聚合语句中，设置：size：0，就会使用分片请求缓存缓存结果。size = 0 的含义是：只返回聚合结果，不返回查询结果。

GET /es_db/_search
{"size": 0, 		 # 添加size： 0"aggs": {"hs_remark_agg": {"terms": {"field": "remark.keyword"}}}
}

拆分聚合，使聚合并行化

Elasticsearch 查询条件中同时有多个条件聚合，默认情况下聚合不是并行运行的。当为每个聚合提供自己的查询并执行 msearch 时，性能会有显著提升。因此，在 CPU 资源不是瓶颈的前提下，如果想缩短响应时间，可以将多个聚合拆分为多个查询，

借助：_msearch 实现并行聚合。

#常规的多条件聚合实现
GET /employees/_search
{"size": 0,"aggs": {"hs_job_agg": {"terms": {"field": "job.keyword"}},"hs_max_salary":{"max": {"field": "salary"}}}
}
# msearch 拆分多个语句的聚合实现
GET _msearch
{"index":"employees"}
{"size":0,"aggs":{"hs_job_agg":{"terms":{"field": "job.keyword"}}}}
{"index":"employees"}
{"size":0,"aggs":{"hs_max_salary":{"max":{"field": "salary"}}}}