利用Rollover Api来进行索引管理(上)

作者: 小周写程序 | 来源:发表于2020-09-10 09:14 被阅读0次

利用Rollover Api来进行索引管理(上)
elasticsearch rollover index
elasticSearch的rollover api（索引声明周
HTML5与 iOS 交互方法总结
H5与Native(iOS)交互
ES Rollover and Shrink API
Elastic Search 索引 API 介绍
python对es基础的增删改查
利用Python脚本通过搜索引擎批量寻找利用注入点
利用zabbix api进行运维管理

Rollover Pattern 索引滚动

为什么要使用滚动索引
比如现在有一个要实时记录用户定位信息的需求，每隔5秒钟就要收集一次。
这时我们将数据存储在ES中。
而我们索引只有一个，伴随着数据越来越多，索引的读写肯定会越来越吃力。
这时我们就可以将索引拆开，将一个索引拆分为多个索引。
这种拆分可以利用索引模板来实现，按照时间创建索引可以使用Rollover 来实现。
rollover使您可以根据索引大小，文档数或使用期限自动过渡到新索引。
当rollover触发后，将创建新索引，写别名（write alias)将更新为指向新索引，所有后续更新都将写入新索引。
注意：对于基于时间的rollover来说，基于大小，文档数或使用期限过渡至新索引是比较适合的。
基于时间的rollover通常会导致许多小的索引，这可能会对性能和资源使用产生负面影响。
对于Rollover后的历史数据，可能已经没有了价值，我们不得不进行删除，但是有一些数据可能对于分析很有用处
ES在6.3之后引入了下面的新功能用来处理历史数据的问题

以紧凑的聚合格式保存旧数据
仅保存您感兴趣的数据这两项功能

2.创建流程

创建索引模板,此时定义模板的别名为zz_user_location_search

curl -XPUT "http://host:port/_template/zz_user_location" -H 'Content-Type: application/json' -d'
{
  "index_patterns": "zz_user_location*",
  "settings": {
    "index":{
      "max_result_window": "1000000"
    },
    "number_of_shards": 3,
    "number_of_replicas": 1
  },
  "mappings": {
    "location": {
      "properties": {
        "jd": {
          "type": "keyword"
        },
        "wd": {
          "type": "keyword"
        },
        "wz": {
          "type": "text",
          "fields": {
            "keyword": {
              "type": "keyword",
              "ignore_above": 256
            }
          }
        },
        "device": {
          "type": "keyword"
        },
        "create_time": {
          "type": "date",
          "format": "yyyy-MM-dd HH:mm:ss||epoch_millis"
        }
      }
    }
  },
  "aliases": {
    "zz_user_location_search": {}
  }
}'

创建索引并给当前索引创建写别名

PUT /zz_user_location_20200909
{
  "aliases": {
    "zz_user_location_save": {}
  }
}

在写别名上执行新增操作

POST /zz_user_location_save/location
{
  "jd":"555"
}

查询索引拥有的别名

GET /zz_user_location_20200909/_alias/* 
--------------result-----------------
{
  "zz_user_location_20200909" : {
    "aliases" : {
      "zz_user_location_save" : { },
      "zz_user_location_search" : { }
    }
  }
}

此时新增数据，两个索引均能查询出数据。比如我们设置当zz_user_location_20200909索引的文档数量大于等于5条或者时间大于等于7天或者索引容量大于等于1个G滚动新索引

dry_run只是测试是否满足条件，并不是真的运行
POST /zz_user_location_save/_rollover/zz_user_location_20200910?dry_run
{
  "conditions": {
    "max_age": "7d",
    "max_docs": 2,
    "max_size": "1gb"
  }
}
-------------result--------------
"dry_run" 是false，证明不满足条件，如果其中一个满足条件，那么就是true了
{
  "acknowledged" : false,
  "shards_acknowledged" : false,
  "old_index" : "zz_user_location_20200909",
  "new_index" : "zz_user_location_20200910",
  "rolled_over" : false,
  "dry_run" : false,
  "conditions" : {
    "[max_size: 1gb]" : false,
    "[max_docs: 2]" : false,
    "[max_age: 7d]" : false
  }
}

如果其中一个满足条件，执行索引滚动

POST zz_user_location_save/_rollover/zz_user_location_20200910
{
  "conditions": {
    "max_age":   "7d",
    "max_docs":  5,
    "max_size": "1gb"
  }
}
-------------result--------------
"rolled_over" 是true，证明滚动成功
{
  "acknowledged" : true,
  "shards_acknowledged" : true,
  "old_index" : "zz_user_location_20200909",
  "new_index" : "zz_user_location_20200910",
  "rolled_over" : true,
  "dry_run" : false,
  "conditions" : {
    "[max_size: 1gb]" : false,
    "[max_docs: 5]" : true,
    "[max_age: 7d]" : false
  }
}

-此时在zz_user_location_save新增一条数据，查询zz_user_location_search,查询的文档分布在不同的索引上

    {
        "_index" : "zz_user_location_20200909",
        "_type" : "location",
        "_id" : "2GCKdXQBHICozJThgsAb",
        "_score" : 1.0,
        "_source" : {
          "jd" : "555"
        }
      },
      {
        "_index" : "zz_user_location_20200910",
        "_type" : "location",
        "_id" : "d2CPdXQBHICozJThG8X0",
        "_score" : 1.0,
        "_source" : {
          "jd" : "555"
        }
      }

创建模板和索引

sh脚本

#!/bin/bash
TIME=$(date "+%Y%m%d")
#ES_URL
ES_URL=$1
#模板前缀
TEMPLATE_PREFIX="zz_user_location"
#初始化写索引名称
INIT_INDEX="${TEMPLATE_PREFIX}_${TIME}"
#写别名
INDEX_SAVE_ALIASE="${TEMPLATE_PREFIX}_save"
#读别名
INDEX_SEARCH_ALIASE="${TEMPLATE_PREFIX}_search"

#获取模板
GET_TEMPLATE_RESULT=$(curl -XGET ${ES_URL}/_template/${TEMPLATE_PREFIX})

#创建模板
createTemplate() {
  RESULT=$(curl -XPUT ${ES_URL}"/_template/${TEMPLATE_PREFIX}" -H 'Content-Type: application/json' -d'
{
  "index_patterns": "'${TEMPLATE_PREFIX}'*",
  "settings": {
    "index": {
      "max_result_window": "1000000"
    },
    "number_of_shards": 3,
    "number_of_replicas": 1
  },
  "mappings": {
    "location": {
      "properties": {
        "user_name": {
          "type": "keyword"
        },
        "jd": {
          "type": "keyword"
        },
        "wd": {
          "type": "keyword"
        },
        "wz": {
          "type": "text",
          "fields": {
            "keyword": {
              "type": "keyword",
              "ignore_above": 256
            }
          }
        },
        "device": {
          "type": "keyword"
        },
        "time": {
          "type": "date",
          "format": "yyyy-MM-dd||epoch_millis"
        },
        "create_time": {
          "type": "date",
          "format": "yyyy-MM-dd HH:mm:ss||epoch_millis"
        }
      }
    }
  },
  "aliases": {
    "'${INDEX_SEARCH_ALIASE}'": {}
  }
}')
  echo "$RESULT"
}

#初始化索引
initTemplateIndex() {
  RESULT=$(curl -XPUT "$ES_URL/${INIT_INDEX}" -H 'Content-Type: application/json' -d'
{
  "aliases": {
    "'${INDEX_SAVE_ALIASE}'": {}
  }
}')
  echo "$RESULT"
}

if test "$GET_TEMPLATE_RESULT" = "{}"; then
  echo '模板未创建！'
  echo '开始创建模板!'
  CREATETEMPLATE_RESULT=$(createTemplate)
  if test "$CREATETEMPLATE_RESULT"="{"acknowledged":true}"; then
    echo '模板创建成功!'
    INITTEMPLATE_RESULT=$(initTemplateIndex)
    echo "result${INITTEMPLATE_RESULT}"
  fi
else
  echo '模板已创建！'
fi

shellutil工具类

public class ShellUtil {
    private static Logger logger = LoggerFactory.getLogger(ShellUtil.class);

    /**
     * 执行shell脚本
     *
     * @param shell
     * @return
     */
    public static List<String> runShell(String shell) {
        Process process;
        List<String> processList = new ArrayList<>();
        try {
            process = Runtime.getRuntime().exec(new String[]{"/bin/sh", "-c", shell});
            try {
                int exitValue = process.waitFor();
                if (0 != exitValue) {
                    logger.error("call shell failed. error code is :" + exitValue);
                    return processList;
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
                return processList;
            }
            BufferedReader input = new BufferedReader(new InputStreamReader(process.getInputStream()));
            String line = "";
            while ((line = input.readLine()) != null) {
                processList.add(line);
            }
            input.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return processList;
    }
}

初始化脚本的方法

 String basePath;
        try {
            basePath = getBasePath();
            String userLocationEsShPath = basePath + File.separator + "bin/userlocation/es_init.sh";
            List<String> list = ShellUtil.runShell(userLocationEsShPath + " " + esSetting.getEsHttpHost());
            for (String s : list) {
                if (s.startsWith("result")) {
                    logger.info("脚本索引创建返回值为{}", s.substring(6));
                    logger.info("{}脚本执行成功", "userlocation");
                }
                logger.info(s);
            }
        } catch (IOException e) {
            logger.error("userlocation脚本执行失败", e);
        }

满足一定条件后执行滚动

sh脚本

#!/bin/bash
TIME=$(date "+%Y%m%d")
#最长时间
max_age=$1
#最大数量
max_docs=$2
#url
url=$3
curl -XPOST "${url}/zz_user_location_save/_rollover/zz_user_location_${TIME}" -H 'Content-Type: application/json' -d'
{
  "conditions": {
    "max_age": "'${max_age}'",
    "max_docs": '${max_docs}'
  }
}'

java定时任务执行方法

try {
            String basePath = new File("").getCanonicalPath();
            String userLocationEsShPath = basePath + File.separator + "bin/userlocation/es_rollover.sh";
            //param1 最长时间 param2 最大数量 param3 ES_URL
            List<String> list = ShellUtil.runShell(userLocationEsShPath + " " + esSetting.getRolloverTime() + " " + esSetting.getRolloverSize() + " " + esSetting.getEsHttpHost());
            if (CollectionUtils.isNotEmpty(list)) {
                String s = list.get(0);
                try {
                    JSONObject object = JSONObject.parseObject(s);
                    if (!object.getBoolean("rolled_over")) {
                        logger.info("userlocation索引未达滚动要求");
                        return;
                    }
                    logger.info("userlocation已滚动至新索引");
                } catch (Exception e) {
                    logger.error("snapshot的返回json解析失败", e);
                }
            }
        } catch (IOException e) {
            logger.error("snapshot脚本执行失败", e);
        }

利用Rollover Api来进行索引管理(上)
Rollover Pattern 索引滚动为什么要使用滚动索引比如现在有一个要实时记录用户定位信息的需求，每隔5...
elasticsearch rollover index
rollover作用当ES索引过大时，rollover，满足条件后，新建索引，将索引别名转向新索引。 rollo...
elasticSearch的rollover api（索引声明周
rollover可以根据索引大小，文档数或使用期限自动过渡到新索引。当rollover触发后，将创建新索引，写别名...
HTML5与 iOS 交互方法总结
1. 利用 WKWebView 进行交互(系统 API) 2. 利用 UIWebView 进行交互(系统 API)...
H5与Native(iOS)交互
主要方式: 1、利用WKWebView进行(系统API)2、利用UIWebView进行(系统API)3、javas...
ES Rollover and Shrink API
Shrink API Shrink API允许我们将存在的索引转变为一个只包含主要分片的新索引。目标索引中请求的...
Elastic Search 索引 API 介绍
前言 Elastic Search 主要靠 Restful api 来对索引进行操作。主要分为两类，一类是索引的管...
python对es基础的增删改查
安装API python对索引进行操作建立es连接创建索引删除索引判断索引存在对索引加入mapping ...
利用Python脚本通过搜索引擎批量寻找利用注入点
思路 : 利用现有的搜索引擎API以及GoogleHacking技术 , 批量进行关键字查询和注入点检测分为三步 ...
利用zabbix api进行运维管理
-- 导读 zabbix 现在之所以那么流行，个人感觉跟zabbix 强大的API有一个很大的关系，利用API...