High cardinality下对持续写入的Elasticse

作者: bellengao | 来源:发表于2019-05-07 15:33 被阅读13次

    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, 字段的唯一值越多(igh 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控制台中,选择"新建",进入云函数创建页面:


    image

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

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


    image

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

    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分钟执行:


    image

    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])
    
    
    def main_handler(event,context):
        delete_old_index()
    
    

    函数说明:

    • 该函数用于删除前一天的按小时建的索引,如当前天是2019-06-07, 则函数执行时会删除
      2019-06-06-00到2019-06-06-23全部24个索引

    配置定期删除索引函数的触发方式为每天的2点执行(SCF使用的是UTC时间,所以cron表达式中需要加8个小时):

    image

    总结

    • 经过以上分析与实战,我们最终降低了High cardinality下对持续写入的Elasticsearch索引进行聚合查询的时延,在利用缓存的情况下,聚合查询响应在ms级
    • 相比按天建索引,采用按小时建索引的优化方案,增加了部分冗余的数据,分片的数量也有增加;因为每小时的数据量相比每天要小的多,所以按小时建的索引分片数量可以设置的低一些,防止出现分片数量过多而大量占用内存的情况
    • 如果数据量比较大,reindex会比较慢,可以通过snapshot api把按小时建的索引数据导入到按天建的索引中,数据导入的速度会比较快,可以参考文档
      https://cloud.tencent.com/document/product/845/19549

    相关文章

      网友评论

        本文标题:High cardinality下对持续写入的Elasticse

        本文链接:https://www.haomeiwen.com/subject/hwaeoqtx.html