High cardinality下对持续写入的Elasticsearch索引进行聚合查询的性能优化

High cardinality下对持续写入的Elasticsearch索引进行聚合查询的性能优化

背景

最近使用腾讯云Elasticsearch Service的用户提出，对线上的ES集群进行查询，响应越来越慢，希望能帮忙优化一下。

查询越来越慢的语句如下：

{
  "_source": false,
  "size": 0,
  "aggs": {
    "traceId": {
      "aggs": {
        "timestamp_millis": {
          "min": {
            "field": "timestamp_millis"
      "terms": {
        "field": "traceId",
        "order": {
          "timestamp_millis": "desc"
        "size": 10
  "query": {
    "bool": {
      "filter": {
        "bool": {
          "must": [
              "range": {
                "timestamp_millis": {
                  "from": 1556431798000,
                  "include_lower": true,
                  "include_upper": true,
                  "to": 1556435398000
              "term": {
                "user": "1275813850"
}

从查询语句上看到用户使用了聚合查询(aggregation query), 第一反应就是聚合查询影响了查询速度。但是又发现，用户的索引是按天创建的，查询昨天的数据量较大的索引(300GB)响应并不慢，可以达到ms级别，但是查询当天的正在写入数据的索引就很慢，并且响应时间随着写入数据的增加而增加。

原因分析

初步分析查询性能瓶颈就在于聚合查询，但是又不清楚为什么查询旧的索引会比较快，而查询正在写入的索引会越来越慢。所以趁机找了些资料了解了下聚合查询的实现，最终了解到：

1. 聚合查询会对要进行聚合的字段构建Global Cardinals, 字段的唯一值越多(high cardinality)，构建Global Cardinals构建的越慢，参考文章： https://blog.csdn.net/zwgdft/article/details/83215977
2. 聚合查询时构建好的Global Cardinals是存放在内存中的，如果索引不再发生变化(没有新数据写入而产生新的segment或者segment merge时)， Global Cardinals就不需要重新构建，第一次进行聚合查询时会构建好Global Cardinals，后续的查询就会使用在内存中已经缓存好的Global Cardinals了
3. 尝试在查询时增加execute_hit：map参数，结果无效，原因是用户使用的6.4.3版本的集群该功能存在bug，虽然通过该参数execute_hit指定了不创建Global Cardinals，但是实际上还是创建了，后续版本已经修复了这个问题, 参考 https://github.com/elastic/elasticsearch/issues/37705

优化方案

经过最终讨论，决定从业务角度对查询性能进行优化，既然对持续写入的索引构建Global Cardinals会越来越慢，那就降低索引的粒度，使得持续写入的索引数据量降低，同时增加了能够使用Global Cardinals缓存的索引数据量。

详细的优化方案如下：

1. 降低索引的粒度，按小时创建索引
2. 写入时只写入当前小时的索引，查询时根据时间范围查询对应的索引
3. 为了防止索引数量和分片数量膨胀，可以把旧的按小时创建的索引定期reindex到一个以当天日期为后缀的索引中，reindex完成之后再删除按小时创建的索引。

实战过程

根据优化方案，需要实现的内容包括：

1. 按小时创建索引，写入数据
2. 每小时执行一次reindex, 把按小时建的索引reindex到按天建的索引中
3. 定期删除按小时建的索引

其中，第一步需要在client端进行，写入数据时根据当前时间指定索引名称，如当前时间是

"2019-05-07 03:50:06", 则写入的索引名称为2019-05-07-03；第二步和第三步都是定时任务，实战时尝试使用SCF(腾讯云Serverless云函数)进行简单的配置即可。

1.创建SCF云函数

在 腾讯云SCF 控制台中，选择"新建"，进入云函数创建页面：

配置函数名称,选择名为"ES写入函数"的模板，该模板自带elasticsearch模块，可以使用es的api操作集群。

创建完成后，需要在"函数配置"TAB页对函数的网络进行配置,选择和Elasticsearch集群同vpc下的网络：

接下来，就可以配置函数代码和触发方式，并进行测试。

###1. 定期reindex

定期reindex的函数代码如下：

# -*- coding: utf8 -*-
from datetime import datetime
from elasticsearch import Elasticsearch
import random
import time
# ES集群地址
ESServer = Elasticsearch("10.0.128.35:9200")
def reindex_hourly_2_daily():
    # 索引前缀，到月份
    index_prefix = "test-index-"+time.strftime( "%Y-%m" ,time.localtime(time.time())) +"-"
    # 当前天
    current_day = time.localtime(time.time()).tm_mday
    # 当前小时，因为SCF是UTC时间，所以加8个小时，如果不在SCF里运行，则不用加8个小时，也不用进行时区转换
    current_hour = time.localtime(time.time()).tm_hour + 8
    # 时区转换
    if current_hour >=24:
        current_hour= current_hour-24
        current_day = current_day +1
    # 前一个小时
    last_hour = current_hour -1
    # 前一天
    last_day = current_day-1
    # 前一个小时的索引
    last_hour_index=''
    # 按天建的索引
    daily_index=''
    # 如果是0点，则把前一天23点的索引迁移到前一天按天建的索引
    if current_hour ==0:
        last_hour=23
        last_day = current_day-1
        # 构造出如2019-05-05格式的索引，日期中的天数小于10则补0
        if last_day<10:
            daily_index = index_prefix +  "0"+ str(last_day)
        else:
            daily_index = index_prefix +  str(last_day)
        # 构造出如2019-05-05-01格式的索引，日期中的小时数小于10则补0
        if last_hour<10:
            last_hour_index = daily_index+  "-0"+ str(last_hour)
        else:
            last_hour_index = daily_index+  "-"+str(last_hour)
    else:
         # 构造出如2019-05-05格式的索引
        if current_day<10:
            daily_index = index_prefix +  "0"+ str(current_day)
        else:
            daily_index = index_prefix +  str(current_day)
        if last_hour<10:
            last_hour_index = daily_index+ "-0"+ str(last_hour)
        else:
            last_hour_index = daily_index+ "-"+ str(last_hour)
    # 自动创建按天建的索引
    ESServer.indices.create(daily_index, ignore=400)
    body= {}
    source ={
        'index':last_hour_index
    dest = {
        'index':daily_index
    body={
        'source':source,
        'dest':dest
    # 执行reindex，source和index相同的情况下，重复执行多次也不会造成数据重复
    rsp = ESServer.reindex(body=body,wait_for_completion=False)
    # 执行reindex返回taskId， 可以通过轮询taskId判断操作是否完成
    print rsp
def main_handler(event,context):
    reindex_hourly_2_daily()

函数代码说明：

• 使用该函数时需要把ES集群地址修改为自己的集群地址
• SCF执行时使用的时间是UTC时间而不是东八区，所以在编写函数代码的时候需要注意进行时区转换
• 调用reindex api时指定wait_for_completion为false, 让reindex操作异步执行，同时返回一个taskId, 后续可以通过task api轮询该task查看任务是否完成；可以选择在reindex完成后删除按小时建的索引, 也可以选择延迟删除，后续定期清理掉按小时建的索引
• 无需担心函数重复执行造成数据重复的情况，reindex执行的是一个upsert操作， 如果source index中的docId在dest index中不存在，则插入该doc,否则更新该doc

配置定期reindex函数的触发方式为每小时的第1分钟执行:

2. 定期删除按小时建的索引

根据需要，可以选择在每天凌晨0点到5点这个时间段，业务请求量不大时，删除前一天按小时建的索引，避免过多的重复数据，以及避免分片数量膨胀。

函数代码如下：

# -*- coding: utf8 -*-
from datetime import datetime
from elasticsearch import Elasticsearch
import random
import time
ESServer = Elasticsearch("10.0.128.35:9200")
def delete_old_index():
    # 索引前缀，到月份
    index_prefix = "test-index-"+time.strftime( "%Y-%m" ,time.localtime(time.time())) +"-"
    # 当前天
    current_day = time.localtime(time.time()).tm_mday
    # 当前小时，因为SCF是UTC时间，所以加8个小时，如果不在SCF里运行，则不用加8个小时，也不用进行时区转换
    current_hour = time.localtime(time.time()).tm_hour + 8
    # 前一天
    last_day = current_day-1
    if current_hour >=24:
        last_day = current_day
    # 需要删除的索引，以通配符表示，如2019-05-05-*，表示删除前一天所有的按小时建的索引
    will_delete_index_prefix=''
    if last_day<10:
        will_delete_index_prefix = index_prefix +  "0"+ str(last_day) +"-"
    else:
        will_delete_index_prefix = index_prefix +  str(last_day)+"-"
    for i in range(24):
        hour = ""
        if i<10:
            hour = "0"+str(i)
        else:
            hour = str(i)
        ESServer.indices.delete(will_delete_index_prefix+hour, ignore=[400, 404])