{

    "query": {

        "match": {

            "title": "{{keywords}}"

        }

    }

}

{

    "query": {

        "match": {

            "overview": "{{keywords}}"

        }

    }

}

movieJudgments = judgments_by_qid(judgments_from_file(filename=JUDGMENTS_FILE))

log_features(es, judgments_dict=movieJudgments, search_index=INDEX_NAME)

build_features_judgments_file(movieJudgments, filename=JUDGMENTS_FILE_FEATURES)

for modelType in [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]:

    # 0, MART

    # 1, RankNet

    # 2, RankBoost

    # 3, AdaRank

    # 4, coord Ascent

    # 6, LambdaMART

    # 7, ListNET

    # 8, Random Forests

    # 9, Linear Regression

    Logger.logger.info("*** Training %s " % modelType)

    train_model(judgments_with_features_file=JUDGMENTS_FILE_FEATURES, model_output='model.txt',

                which_model=modelType)

    save_model(script_name="test_%s" % modelType, feature_set=FEATURE_SET_NAME,       model_fname='model.txt')

{

  "query": {

      "multi_match": {

          "query": "Rambo",

          "fields": ["title", "overview"]

       }

   },

  "rescore": {

      "query": {

        "rescore_query": {

            "sltr": {

                "params": {

                    "keywords": "Rambo"

                },

                "model": "test_1",

            }

         }

      }

   }

}

ELK 从发布5.0之后加入了beats套件之后，就改名叫做elastic stack了。beats是一组轻量级的软件，给我们提供了简便，快捷的方式来实时收集、丰富更多的数据用以支撑我们的分析。但由于beats都需要安装在ELK集群之外，在宿主机之上，其对宿主机的性能的影响往往成为了考量其是否能被使用的关键，而不是它到底提供了什么样的功能。因为业务的稳定运行才是核心KPI，而其他因运维而生的数据永远是更低的优先级。影响宿主机性能的方面可能有很多，比如CPU占用率，网络吞吐占用率，磁盘IO，内存等，这里我们详细讨论一下内存泄漏的问题

@[toc]

filebeat是beats套件的核心组件之一（另一个核心是metricbeat），用于采集文件内容并发送到收集端（ES），它一般安装在宿主机上，即生成文件的机器。根据文档的描述，filebeat是不建议用来采集NFS（网络共享磁盘）上的数据的，因此，我们这里只讨论filebeat对本地文件进行采集时的性能情况。

当filebeat部署和运行之后，必定会对cpu，内存，网络等资源产生一定的消耗，当这种消耗能够限定在一个可接受的范围时，在企业内部的生产服务器上大规模部署filebeat是可行的。但如果出现一些非预期的情况，比如占用了大量的内存，那么运维团队肯定是优先保障核心业务的资源，把filebeat进程给杀了。很可惜的是，内存泄漏的问题，从filebeat的诞生到现在就一直没有完全解决过。（可以区社区讨论贴看看，直到现在V6.5.1都还有人在报告内存泄漏的问题）。在特定的场景和配置下，内存占用过多已经成为了抑止filebeat大规模部署的主要问题了。在这里，我主要描述一下我碰到的在filebeat 6.0上遇到的问题。

问题场景和配置

一开始我们在很多机器上部署了filebeat，并且使用了一套统一无差别的的简单配置。对于想要在企业内部大规模推广filebeat的同学来说，这是大忌！！！ 合理的方式是具体问题具体分析，需对每台机器上产生文件的方式和rotate的方式进行充分的调研，针对不同的场景是做定制化的配置。以下是我们之前使用的配置：

• multiline，多行的配置，当日志文件不符合规范，大量的匹配pattern的时候，会造成内存泄漏
• max_procs，限制filebeat的进程数量，其实是内核数，建议手动设为1

filebeat.prospectors:
- type: log
  enabled: true
  paths:
    - /qhapp/*/*.log
  tail_files: true
  multiline.pattern: '^[[:space:]]+|^Caused by:|^.+Exception:|^\d+\serror'
  multiline.negate: false
  multiline.match: after
  fields:
    app_id: bi_lass
    service: "{{ hostvars[inventory_hostname]['service'] }}"
    ip_address: "{{ hostvars[inventory_hostname]['ansible_host'] }}"
    topic: qh_app_raw_log

filebeat.config.modules:
  path: ${path.config}/modules.d/*.yml
  reload.enabled: false

setup.template.settings:
  index.number_of_shards: 3
  #index.codec: best_compression
  #_source.enabled: false
output.kafka:
  enabled: true
  hosts: [{{kafka_url}}]

  topic: '%{[fields][topic]}'

max_procs: 1

注意，以上的配置中，仅仅对cpu的内核数进行了限制，而没有对内存的使用率进行特殊的限制。从配置层面来说，影响filebeat内存使用情况的指标主要有两个：

• queue.mem.events消息队列的大小，默认值是4096，这个参数在6.0以前的版本是spool-size，通过命令行，在启动时进行配置
• max_message_bytes 单条消息的大小, 默认值是10M

filebeat最大的可能占用的内存是max_message_bytes * queue.mem.events = 40G，考虑到这个queue是用于存储encode过的数据，raw数据也是要存储的，所以，在没有对内存进行限制的情况下，最大的内存占用情况是可以达到超过80G。

因此，建议是同时对filebeat的CPU和内存进行限制。

下面，我们看看，使用以上的配置在什么情况下会观测到内存泄漏

监控文件过多

对于实时大量产生内容的文件，比如日志，常用的做法往往是将日志文件进行rotate，根据策略的不同，每隔一段时间或者达到固定大小之后，将日志rotate。 这样，在文件目录下可能会产生大量的日志文件。 如果我们使用通配符的方式，去监控该目录，则filebeat会启动大量的harvester实例去采集文件。但是，请记住，我这里不是说这样一定会产生内存泄漏，只是在这里观测到了内存泄漏而已，不是说这是造成内存泄漏的原因。

当filebeat运行了几个月之后，占用了超过10个G的内存

非常频繁的rotate日志

另一个可能是，filebeat只配置监控了一个文件，比如test2.log，但由于test2.log不停的rotate出新的文件，虽然没有使用通配符采集该目录下的所有文件，但因为linux系统是使用inode number来唯一标示文件的，rotate出来的新文件并没有改变其inode number，因此，时间上filebeat还是同时开启了对多个文件的监控。

另外，因为对文件进行rotate的时候，一般会限制rotate的个数，即到达一定数量时，新rotate一个文件，必然会删除一个旧的文件，文件删除之后，inode number是可以复用的，如果不巧，新rotate出来的文件被分配了一个之前已删掉文件的inode number，而此时filebeat还没有监测之前持有该inode number的文件已删除，则会抛出以下异常：

2018-11-21T18:06:55+08:00 ERR  Harvester could not be started on truncated file: /qhapp/logs/bd-etl/logs/test2.log, Err: Error setting up harvester: Harvester setup failed. Unexpected file opening error: file info is not identical with opened file. Aborting harvesting and retrying file later again

而类似Harvester setup failed.的异常会导致内存泄漏

https://github.com/elastic/beats/issues/6797

因为multiline导致内存占用过多

multiline.pattern: '^[[:space:]]+|^Caused by:|^.+Exception:|^\d+\serror，比如这个配置，认为空格或者制表符开头的line是上一行的附加内容，需要作为多行模式，存储到同一个event当中。当你监控的文件刚巧在文件的每一行带有一个空格时，会错误的匹配多行，造成filebeat解析过后，单条event的行数达到了上千行，大小达到了10M，并且在这过程中使用的是正则表达式，每一条event的处理都会极大的消耗内存。因为大多数的filebeat output是需应答的，buffer这些event必然会大量的消耗内存。

模拟场景

这里不多说，简单来一段python的代码：

[loggers]
keys=root

[handlers]
keys=NormalHandler

[formatters]
keys=formatter

[logger_root]
level=DEBUG
handlers=NormalHandler

[handler_NormalHandler]
class=logging.handlers.TimedRotatingFileHandler
formatter=formatter
args=('./test2.log', 'S', 10, 200)

[formatter_formatter]
format=%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s

以上，每隔10秒（'S', 'M' = 分钟，'D'= 天）rotate一个文件，一共可以rotate 200个文件。 然后，随便找一段日志，不停的打，以下是330条/秒

import logging
from logging.config import fileConfig
import os
import time
CURRENT_FOLDER = os.path.dirname(os.path.realpath(__file__))

fileConfig(CURRENT_FOLDER + '/logging.ini')
logger = logging.getLogger()

while True:
    logger.debug("DEBUG 2018-11-26 09:31:35 com.sunyard.insurance.server.GetImage 43 - 资源请求:date=20181126&file_name=/imagedata/imv2/pool1/images/GXTB/2017/11/14/57/06b6bcdd31763b70b20f56c689e51f5e_1/06b6bcdd31763b70b20f56c689e51f5e_2.syd&file_encrypt=0&token=HUtGGG20GH4GAqq209R9tc9UGtAURR0b DEBUG 2018-11-26 09:31:40 com.sunyard.insurance.scheduler.job.DbEroorHandleJob 26 - [数据库操作异常处理JOB]处理异常文件，本机不运行，退出任务!")
    logger.debug("DEBUG 2018-11-26 09:31:35 com.sunyard.insurance.server.GetImage 43 - 资源请求:date=20181126&file_name=/imagedata/imv2/pool1/images/GXTB/2017/11/14/57/06b6bcdd31763b70b20f56c689e51f5e_1/06b6bcdd31763b70b20f56c689e51f5e_2.syd&file_encrypt=0&token=HUtGGG20GH4GAqq209R9tc9UGtAURR0b DEBUG 2018-11-26 09:31:40 com.sunyard.insurance.scheduler.job.DbEroorHandleJob 26 - [数据库操作异常处理JOB]处理异常文件，本机不运行，退出任务!")
    logger.debug("DEBUG 2018-11-26 09:31:35 com.sunyard.insurance.server.GetImage 43 - 资源请求:date=20181126&file_name=/imagedata/imv2/pool1/images/GXTB/2017/11/14/57/06b6bcdd31763b70b20f56c689e51f5e_1/06b6bcdd31763b70b20f56c689e51f5e_2.syd&file_encrypt=0&token=HUtGGG20GH4GAqq209R9tc9UGtAURR0b DEBUG 2018-11-26 09:31:40 com.sunyard.insurance.scheduler.job.DbEroorHandleJob 26 - [数据库操作异常处理JOB]处理异常文件，本机不运行，退出任务!")
    logger.debug("DEBUG 2018-11-26 09:31:35 com.sunyard.insurance.server.GetImage 43 - 资源请求:date=20181126&file_name=/imagedata/imv2/pool1/images/GXTB/2017/11/14/57/06b6bcdd31763b70b20f56c689e51f5e_1/06b6bcdd31763b70b20f56c689e51f5e_2.syd&file_encrypt=0&token=HUtGGG20GH4GAqq209R9tc9UGtAURR0b DEBUG 2018-11-26 09:31:40 com.sunyard.insurance.scheduler.job.DbEroorHandleJob 26 - [数据库操作异常处理JOB]处理异常文件，本机不运行，退出任务!")
    logger.debug("DEBUG 2018-11-26 09:31:35 com.sunyard.insurance.server.GetImage 43 - 资源请求:date=20181126&file_name=/imagedata/imv2/pool1/images/GXTB/2017/11/14/57/06b6bcdd31763b70b20f56c689e51f5e_1/06b6bcdd31763b70b20f56c689e51f5e_2.syd&file_encrypt=0&token=HUtGGG20GH4GAqq209R9tc9UGtAURR0b DEBUG 2018-11-26 09:31:40 com.sunyard.insurance.scheduler.job.DbEroorHandleJob 26 - [数据库操作异常处理JOB]处理异常文件，本机不运行，退出任务!")
    logger.debug("DEBUG 2018-11-26 09:31:35 com.sunyard.insurance.server.GetImage 43 - 资源请求:date=20181126&file_name=/imagedata/imv2/pool1/images/GXTB/2017/11/14/57/06b6bcdd31763b70b20f56c689e51f5e_1/06b6bcdd31763b70b20f56c689e51f5e_2.syd&file_encrypt=0&token=HUtGGG20GH4GAqq209R9tc9UGtAURR0b DEBUG 2018-11-26 09:31:40 com.sunyard.insurance.scheduler.job.DbEroorHandleJob 26 - [数据库操作异常处理JOB]处理异常文件，本机不运行，退出任务!")
    logger.debug("DEBUG 2018-11-26 09:31:35 com.sunyard.insurance.server.GetImage 43 - 资源请求:date=20181126&file_name=/imagedata/imv2/pool1/images/GXTB/2017/11/14/57/06b6bcdd31763b70b20f56c689e51f5e_1/06b6bcdd31763b70b20f56c689e51f5e_2.syd&file_encrypt=0&token=HUtGGG20GH4GAqq209R9tc9UGtAURR0b DEBUG 2018-11-26 09:31:40 com.sunyard.insurance.scheduler.job.DbEroorHandleJob 26 - [数据库操作异常处理JOB]处理异常文件，本机不运行，退出任务!")
    logger.debug("DEBUG 2018-11-26 09:31:35 com.sunyard.insurance.server.GetImage 43 - 资源请求:date=20181126&file_name=/imagedata/imv2/pool1/images/GXTB/2017/11/14/57/06b6bcdd31763b70b20f56c689e51f5e_1/06b6bcdd31763b70b20f56c689e51f5e_2.syd&file_encrypt=0&token=HUtGGG20GH4GAqq209R9tc9UGtAURR0b DEBUG 2018-11-26 09:31:40 com.sunyard.insurance.scheduler.job.DbEroorHandleJob 26 - [数据库操作异常处理JOB]处理异常文件，本机不运行，退出任务!")
    logger.debug("DEBUG 2018-11-26 09:31:35 com.sunyard.insurance.server.GetImage 43 - 资源请求:date=20181126&file_name=/imagedata/imv2/pool1/images/GXTB/2017/11/14/57/06b6bcdd31763b70b20f56c689e51f5e_1/06b6bcdd31763b70b20f56c689e51f5e_2.syd&file_encrypt=0&token=HUtGGG20GH4GAqq209R9tc9UGtAURR0b DEBUG 2018-11-26 09:31:40 com.sunyard.insurance.scheduler.job.DbEroorHandleJob 26 - [数据库操作异常处理JOB]处理异常文件，本机不运行，退出任务!")
    logger.debug("DEBUG 2018-11-26 09:31:35 com.sunyard.insurance.server.GetImage 43 - 资源请求:date=20181126&file_name=/imagedata/imv2/pool1/images/GXTB/2017/11/14/57/06b6bcdd31763b70b20f56c689e51f5e_1/06b6bcdd31763b70b20f56c689e51f5e_2.syd&file_encrypt=0&token=HUtGGG20GH4GAqq209R9tc9UGtAURR0b DEBUG 2018-11-26 09:31:40 com.sunyard.insurance.scheduler.job.DbEroorHandleJob 26 - [数据库操作异常处理JOB]处理异常文件，本机不运行，退出任务!!@#!@#!@#!@#!@#!@#!@#!@#!@#!@#!@#!#@!!!@##########################################################################################################################################################")
    time.sleep(0.03)

如何观察filebeat的内存

在6.3版本之前，我们是无法通过xpack的monitoring功能来观察beats套件的性能的。因此，这里讨论的是没有monitoring时，我们如何去检测filebeat的性能。当然，简单的方法是通过top,ps等操作系统的命令进行查看，但这些都是实时的，无法做趋势的观察，并且都是进程级别的，无法看到filebeat内部的真是情况。因此，这里介绍如何通过filebeat的日志和pprof这个工具来观察内存的使用情况

通过filebeat的日志

filebeat文件解读

其实filebeat的日志，已经包含了很多参数用于实时观测filebeat的资源使用情况，以下是filebeat的一个日志片段（这里的日志片段是6.0版本的，6.3版本之后，整个日志格式变了，从kv格式变成了json对象格式，xpack可以直接通过日志进行filebeat的monitoring）：

2018-11-02T17:40:01+08:00 INFO Non-zero metrics in the last 30s: beat.memstats.gc_next=623475680 beat.memstats.memory_alloc=391032232 beat.memstats.memory_total=155885103371024 filebeat.events.active=-402 filebeat.events.added=13279 filebeat.events.done=13681 filebeat.harvester.closed=1 filebeat.harvester.open_files=7 filebeat.harvester.running=7 filebeat.harvester.started=2 libbeat.config.module.running=0 libbeat.output.events.acked=13677 libbeat.output.events.batches=28 libbeat.output.events.total=13677 libbeat.outputs.kafka.bytes_read=12112 libbeat.outputs.kafka.bytes_write=1043381 libbeat.pipeline.clients=1 libbeat.pipeline.events.active=0 libbeat.pipeline.events.filtered=4 libbeat.pipeline.events.published=13275 libbeat.pipeline.events.total=13279 libbeat.pipeline.queue.acked=13677 registrar.states.cleanup=1 registrar.states.current=8 registrar.states.update=13681 registrar.writes=28

里面的参数主要分成三个部分：

• beat.*，包含memstats.gc_next，memstats.memory_alloc，memstats.memory_total，这个是所有beat组件都有的指标，是filebeat继承来的，主要是内存相关的，我们这里特别关注memstats.memory_alloc，alloc的越多，占用内存越大
• filebeat.*，这部分是filebeat特有的指标，通过event相关的指标，我们知道吞吐，通过harvester，我们知道正在监控多少个文件，未消费event堆积的越多，havester创建的越多，消耗内存越大
• libbeat.*，也是beats组件通用的指标，包含outputs和pipeline等信息。这里要主要当outputs发生阻塞的时候，会直接影响queue里面event的消费，造成内存堆积
• registrar，filebeat将监控文件的状态放在registry文件里面，当监控文件非常多的时候，比如10万个，而且没有合理的设置close_inactive参数，这个文件能达到100M，载入内存后，直接占用内存

filebeat日志解析

当然，我们不可能直接去读这个日志，既然我们使用ELK，肯定是用ELK进行解读。因为是kv格式，很方便，用logstash的kv plugin：

filter {
  kv {}
}

kv无法指定properties的type，所以，我们需要稍微指定了一下索引的模版：

PUT _template/template_1
{
  "index_patterns": ["filebeat*"],
  "settings": {
    "number_of_shards": 1
  },
  "mappings": {
    "doc": {
      "_source": {
        "enabled": false
      },
      "dynamic_templates": [
        {
          "longs_as_strings": {
            "match_mapping_type": "string",
            "path_match":   "*beat.*",
            "path_unmatch": "*.*name",
            "mapping": {
              "type": "long"
            }
          }
        }
      ]
    }
  }
}

上面的模版，将kv解析出的properties都mapping到long类型，但注意"path_match": "*beat.*"无法match到registrar的指标，读者可以自己写一个更完善的mapping。 这样，我们就可以通过kibana可视化组件，清楚的看到内存泄漏的过程

以及资源的使用情况：

将信息可视化之后，我们可以明显的发现，内存的突变和ERR是同时发生的

即以下error： 2018-11-27T09:05:44+08:00 ERR Harvester could not be started on new file: /qhapp/logs/bd-etl/logs/test2.log, Err: Error setting up harvester: Harvester setup failed. Unexpected file opening error: file info is not identical with opened file. Aborting harvesting and retrying file later again

会导致filebeat突然申请了额外的内存。具体请查看issue

通过pprof

众所周知，filebeat是用go语言实现的，而go语言本身的基础库里面就包含pprof这个功能极其强大的性能分析工具，只是这个工具是用于debug的，在正常模式下，filebeat是不会启动这个选贤的，并且很遗憾，在官方文档里面根本没有提及我们可以使用pprof来观测filebeat。我们接下来可以通过6.3上修复的一个内存泄漏的issue，来学习怎么使用pprof进行分析

启动pprof监测

首先，需要让filebeat在启动的时候运行pprof，具体的做法是在启动是加上参数-httpprof localhost:6060，即/usr/share/filebeat/bin/filebeat -c /etc/filebeat/filebeat.yml -path.home /usr/share/filebeat -path.config /etc/filebeat -path.data /var/lib/filebeat -path.logs /var/log/filebeat -httpprof localhost:6060。这里只绑定了localhost，无法通过远程访问，如果想远程访问，应该使用0.0.0.0。 这时，你就可以通过curl http://localhost:6060/debug/pprof/heap > profile.txt等命令，获取filebeat的实时堆栈信息了。

远程连接

当然，你也可以通过在你的本地电脑上安装go，然后通过go tool远程连接pprof。 因为我们是需要研究内存的问题，所以以下连接访问的是/heap子路径 go tool pprof http://10.60.x.x:6060/debug/pprof/heap

top 命令

连接之后，你可以通过top命令，查看消耗内存最多的几个实例：

33159.58kB of 33159.58kB total (  100%)
Dropped 308 nodes (cum <= 165.80kB)
Showing top 10 nodes out of 51 (cum >= 512.04kB)
      flat  flat%   sum%        cum   cum%
19975.92kB 60.24% 60.24% 19975.92kB 60.24%  runtime.malg
 7680.66kB 23.16% 83.40%  7680.66kB 23.16%  github.com/elastic/beats/filebeat/channel.SubOutlet
 2048.19kB  6.18% 89.58%  2048.19kB  6.18%  github.com/elastic/beats/filebeat/prospector/log.NewHarvester
 1357.91kB  4.10% 93.68%  1357.91kB  4.10%  runtime.allgadd
 1024.08kB  3.09% 96.76%  1024.08kB  3.09%  runtime.acquireSudog
  544.67kB  1.64% 98.41%   544.67kB  1.64%  github.com/elastic/beats/libbeat/publisher/queue/memqueue.NewBroker
  528.17kB  1.59%   100%   528.17kB  1.59%  regexp.(*bitState).reset
         0     0%   100%   528.17kB  1.59%  github.com/elastic/beats/filebeat/beater.(*Filebeat).Run
         0     0%   100%   512.04kB  1.54%  github.com/elastic/beats/filebeat/channel.CloseOnSignal.func1
         0     0%   100%   512.04kB  1.54%  github.com/elastic/beats/filebeat/channel.SubOutlet.func1

查看堆栈调用图

输入web命令，会生产堆栈调用关系的svg图，在这个svg图中，你可以结合top命令一起查看，在top中，我们已经知道github.com/elastic/beats/filebeat/channel.SubOutlet占用了很多的内存，在图中，展现的是调用关系栈，你可以看到这个类是怎么被实例化的，并且在整个堆中，内存是怎么分布的。最直观的是，实例所处的长方形面积越大，代表占用的内存越多。：

查看源码

通过list命令，可以迅速查看可以实例的问题源码，比如在之前的top10命令中，我们已经看到github.com/elastic/beats/filebeat/channel.SubOutlet这个类的实例占用了大量的内存，我们可以通过list做进一步的分析，看看这个类内部在哪个语句开始出现内存的占用：

(pprof) list SubOutlet
Total: 32.38MB
ROUTINE ======================== github.com/elastic/beats/filebeat/channel.SubOutlet in /home/jeremy/src/go/src/github.com/elastic/beats/filebeat/channel/util.go
    7.50MB     7.50MB (flat, cum) 23.16% of Total
         .          .     15:// SubOutlet create a sub-outlet, which can be closed individually, without closing the
         .          .     16:// underlying outlet.
         .          .     17:func SubOutlet(out Outleter) Outleter {
         .          .     18:   s := &subOutlet{
         .          .     19:       isOpen: atomic.MakeBool(true),
       1MB        1MB     20:       done:   make(chan struct{}),
       2MB        2MB     21:       ch:     make(chan *util.Data),
    4.50MB     4.50MB     22:       res:    make(chan bool, 1),
         .          .     23:   }
         .          .     24:
         .          .     25:   go func() {
         .          .     26:       for event := range s.ch {
         .          .     27:           s.res <- out.OnEvent(event) 

如何调优

其实调优的过程就是调整参数的过程，之前说过了，和内存相关的参数, max_message_bytes,queue.mem.events,queue.mem.flush.min_events，以及队列占用内存的公式:max_message_bytes * queue.mem.events

output.kafka:
  enabled: true
#  max_message_bytes: 1000000
  hosts: ["10.60.x.x:9092"]
  topic: '%{[fields][topic]}'
max_procs: 1 
#queue.mem.events: 256
#queue.mem.flush.min_events: 128

但其实，不同的环境下，不同的原因都可能会造成filebeat占用的内存过大，此时，需要仔细的确认你的上下文环境：

• 是否因为通配符的原因，造成同时监控数量巨大的文件，这种情况应该避免用通配符监控无用的文件。
• 是否文件的单行内容巨大，确定是否需要改造文件内容，或者将其过滤
• 是否过多的匹配了multiline的pattern，并且多行的event是否单条体积过大。这时，就需要暂时关闭multiline，修改文件内容或者multiline的pattern。
• 是否output经常阻塞，event queue里面总是一直缓存event。这时要检查你的网络环境或者消息队列等中间件是否正常

近两年随着Elastic Stack的愈发火热，其近乎成为构建实时日志应用的工业标准。在小型数据应用场景，最新的6.5版本已经可以做到开箱即用，无需过多考虑架构上的设计工作。 然而一旦应用规模扩大到数百TB甚至PB的数据量级，整个系统的架构和后期维护工作则显得非常重要。借着2018 Elastic Advent写文的机会，结合过去几年架构和运维公司日志集群的实践经验，对于大规模日志型数据的管理策略，在此做一个总结性的思考。 文中抛出的观点，有些已经在我们的集群中有所应用并取得比较好的效果，有些则还待实践的检验。抛砖引玉，不尽成熟的地方，还请社区各位专家指正。

对于日志系统，最终用户通常有以下几个基本要求:

1. 数据从产生到可检索的实时性要求高，可接受的延迟通常要求控制在数秒，至多不超过数十秒
2. 新鲜数据(当天至过去几天)的查询和统计频率高，返回速度要快(毫秒级，至多几秒)
3. 历史数据保留得越久越好。

针对这些需求，加上对成本控制的必要性，大家通常想到的第一个架构设计就是冷热数据分离。

冷热数据分离

冷热分离的概念比较好理解，热结点做数据的写入，保存近期热数据，冷数据定期迁移到冷数据结点，就这么简单。不过实际操作起来可能还是碰到一些具体需要考虑的细节问题。

1. 冷热结点集群配置的JVM heap配置要差异化。热结点无需存放太多数据，对于heap的要求通常不是太高，在够用的情况下尽量配置小一点。可以配置在26GB左右甚至更小，而不是大多数人知道的经验值31GB。原因在于这个size的heap，可以启用zero based Compressed Oops，JVM运行效率是最高的， 参考: heap-size。而冷结点存在的目的是尽量放更多的数据，性能不是首要的，因此heap可以配置在31GB。
2. 数据迁移过程有一定资源消耗，为避免对数据写入产生显著影响，通常定时在业务低峰期，日志产出量比较低的时候进行，比如半夜。
3. 索引是否应该启用压缩，如何启用？最初我们对于热结点上的索引是不启用压缩的，为了节省CPU消耗。只在冷结点配置里，增加了索引压缩选项。这样索引迁移到冷结点后，执行force merge操作的时候，ES会自动将结点上配置的索引压缩属性套用到merge过后新生成的segment，这样就实现了热结点不压缩，冷结点merge过后压缩的功能。极大节省了冷结点的磁盘空间。后来随着硬件的升级，我们发现服务器的cpu基本都是过剩的，磁盘IO通常先到瓶颈。 因此尝试在热结点上一开始创建索引的时候，就启用压缩选项。实际对比测试并没有发现显著的索引吞吐量下降，并且因为索引压缩后磁盘文件size的大幅减少，每天夜间的数据迁移工作可以节省大量的时间。至此我们的日志集群索引默认就是压缩的。
4. 冷结点上留做系统缓存的内存一般不多，加上数据量非常巨大。索引默认的mmapfs读取方式，很容易因为系统缓存不够，导致数据在内存和磁盘之间频繁换入换出。严重的情况下，整个结点甚至会因为io持续在100%无法响应。 实践中我们发现对冷结点使用niofs效果会更好。

实现了冷热结点分离以后，集群的资源利用率提升了不少，可管理性也要好很多了. 但是随着接入日志的类型越来越多（我们生产上有差不多400种类型的日志)，各种日志的速率差异又很大，让ES自己管理shard的分布很容易产生写入热点问题。 针对这个问题，可以采用对集群结点进行分组管理的策略来解决。

热结点分组管理

所谓分组管理，就是通过在结点的配置文件中增加自定义的标签属性，将服务器区分到不同的组别中。然后通过设置索引的index.routing.allocation.include属性，控制改索引分布在哪个组别。同时配合设置index.routing.allocation.total_shards_per_node，可以做到某个索引的shard在某个group的结点之间绝对均匀分布。

比如一个分组有10台机器，对一个5 primary ，1 replica的索引，让该索引分布在该分组的同时，设置total_shards_per_node为1，让每个节点上只能有一个分片，这样就避免了写入热点问题。 该方案的缺陷也显而易见，一旦有结点挂掉，不会自动recovery，某个shard将一直处于unassigned状态，集群状态变成yellow。 但我认为，热数据的恢复开销是非常高的，与其立即在其他结点开始复制，之后再重新rebalance，不如就让集群暂时处于yellow状态。 通过监控报警的手段，及时通知运维人员解决结点故障。 待故障解决之后，直接从恢复后的结点开始数据复制，开销要低得多。

在我们的生产环境主要有两种类型的结点分组，分别是10台机器一个分组，和2台机器一个分组。10台机器的分组用于应对速率非常高，shard划分比较多的索引，2台机器的分组用于速率很低，一个shard（加一个复制片）就可以应对的索引。

这种分组策略在我们的生产环境中经过验证，非常好的解决了写入热点问题。那么冷数据怎么管理？ 冷数据不做写入，不存在写入热点问题，查询频率也比较低，业务需求方面对查询响应要求也不那么严苛，所以查询热点问题也不是那么突出。因此为了简化管理，冷结点我们是不做shard分布的精细控制，所有数据迁移到冷数据结点之后，由ES默认的shard分布则略去控制数据的分布。

不过如果想进一步提高冷数据结点服务器资源的利用率，还是可以有进一步挖掘的的空间。我们知道ES默认的shard分布策略，只是保证一个索引的shard尽量分布在不同的结点，同时保证每个节点上shard数量差不多。但是如果采用默认按天创建索引的策略，由于索引速率差异很大，不同索引之间shard的大小差异可能是1-2个数量级的。如果每个shard的size差异不大就好了，那么默认的分布策略，基本上可以保证冷结点之间数据量分布的大致均匀。 能实现类似功能的是ES的rollover特性。

索引的Rollover

Rollover api可以让索引根据预先定义的时间跨度，或者索引大小来自动切分出新索引，从而将索引的大小控制在计划的范围内。合理的应用rollver api可以保证集群shard大小差别不会太大。 只是集群索引类别比较多的时候，rollover全部手动管理负担比较大，需要借助额外的管理工具和监控工具。我们出于管理简便的考虑，暂时没有应用到这个特性。

索引的Rollup

我们发现生产有些用户写入的“日志”，实际上是多维的metrcis数据，使用的时候不是为了查询日志的详情，仅仅是为了做各种维度组合的过滤和聚合。对于这种类型的数据，保留历史数据过多一来浪费存储空间，二来每次聚合都要在裸数据上跑，非常浪费资源。 ES从6.3开始，在x-pack里推出了rollup api，通过定期对裸数据做预先聚合，大大缩减了保存在磁盘上的数据量。对于不需要查询裸日志的应用场景，合理应用该特性，可以将历史数据的磁盘消耗降低几个数量级，同时rollup search也可以大大提升聚合速度。不过rollup也有其局限性，即他的实现是通过定期任务，对间隔期数据跑聚合完成的，有一定的计算开销。 如果数据写入速率非常高，集群压力很大，rollup可能无法跟上写入速率，而不具有实用性。 所以实际环境中，还是需要根据应用场景和资源使用情况，进行灵活的取舍。

多集群的便利性

数据量大到一定程度以后，单集群由于master node单点的限制，会遇到各种集群状态数据更新时得性能问题。 由此现在一些大规模的应用已经开始利用到多集群互联和cross cluster search的特性。 这种结构除了解决单集群数据容量限制问题以外，我们还发现在做容量均衡方面还有比较好的便利性。应用日志写入量通常随着业务变化也会剧烈变化，好不容易规划好的容量，不久就被业务的增长给打破，数倍或者数10倍的流量增长很可能就让一组结点过载出现写入延迟。 如果只有一个集群，在结点之间重新平衡shard比较费力，涉及到数据的迁移，可能非常缓慢，还会影响写入。 但如果有多集群互联，切换就可以做到非常的快速和简单。 原理上只需要在新集群中加入对应的索引配置模版，然后更新写入程序的配置，写入目标指向新集群，重启写入程序即可。并且，可以进一步将整个流程工具化，在GUI上完成一键切换。

一、活动介绍

OpenResty 是一个基于 Nginx 与 Lua 的高性能 Web 平台，其内部集成了大量精良的 Lua 库、第三方模块以及大多数的依赖项。用于方便地搭建能够处理超高并发、扩展性极高的动态 Web 应用、Web 服务和动态网关。目前 OpenResty 是全球占有率排名第五的 Web 服务器，超过 1 万多家商业公司都在使用 OpenResty，比如又拍云、腾讯、奇虎360、京东、12306、Kong 等公司，应用在 CDN、广告系统、搜索、实现业务逻辑、余票查询、网关、ingress controller 等各种不同的场景下。

又拍云 Open Talk 是由又拍云发起的系列主题分享沙龙，秉承又拍云帮助企业提升发展速度的初衷，从 2015 年开启以来，Open Talk 至今已成功举办 45 期，辐射线上线下近十万技术人群，分别在北京、上海、广州、深圳、杭州等 12 座城市举办，覆盖腾讯、华为、网易、京东、唯品会、哔哩哔哩、美团点评、有赞、无码科技等诸多知名企业，往期的活动的讲稿及视频详见：https://opentalk.upyun.com。

为促进 OpenResty 在技术圈的发展，增进 OpenResty 使用者的交流与学习，OpenResty 中国社区联合又拍云，举办 “OpenResty × Open Talk” 全国巡回沙龙，邀请业内资深的 OpenResty 技术专家，分享 OpenResty 实战经验，推动 OpenResty 开源项目的发展，促进互联网技术的教育。

二、活动时间

2019 年 01 月 12 日（周六）

三、活动地址

广东省深圳市南山区深圳虚拟大学园 R3-B 栋一楼触梦社区

四、报名地址

http://www.huodongxing.com/event/5470242889300

五、嘉宾及议题介绍

分享嘉宾一：温铭 / OpenResty 软件基金会主席

OpenResty 软件基金会主席，前360开源委员会委员；现为 OpenResty Inc. 合伙人，致力于打造基于 OpenResty、Linux 内核等基础平台的企业级产品和解决方案。创业之前在互联网安全公司工作了 10 年，主要从事服务端的开发和架构，负责开发过木马云查杀、反钓鱼系统和企业安全产品。

分享主题：《 使用 OpenResty 实现 memcached server 》

OpenResty 软件基金会主席，前360开源委员会委员；现为 OpenResty Inc. 合伙人，致力于打造基于 OpenResty、Linux 内核等基础平台的企业级产品和解决方案。创业之前在互联网安全公司工作了 10 年，主要从事服务端的开发和架构，负责开发过木马云查杀、反钓鱼系统和企业安全产品。

分享嘉宾二：张聪 / 又拍云首席架构师

多年 CDN 行业产品设计、技术开发和团队管理相关经验，个人技术方向集中在 Nginx、OpenResty 等高性能 Web 服务器方面，国内 OpenResty 技术早期推广者之一；目前担任又拍云内容加速部技术负责人，主导又拍云 CDN 技术平台的建设和发展。

分享主题：《 OpenResty 动态流控的几种姿势 》

本次分享主要介绍 OpenResty 上几种常用的请求限制和流量整形的方式，特别值得一提的是，详细介绍在又拍云网关层的实际应用案例，以及又拍云贡献的开源 Resty 库。

分享嘉宾三：张波 / 虎牙直播运维研发架构师

目前主要负责虎牙直播运维体系的建设，主要针对 web 和后台类程序的发布、监控、运维自动化相关的运维系统的设计和开发。

分享主题：《 掘金 Nginx 日志 》

Nginx 是现在最流行的负载均衡和反向代理服务器之一，如果你是一位中小微型网站的开发运维人员，可能仅 Nginx 每天就会产生上百 M 甚至数以十 G 的日志文件，Nginx 日志被删除以前，或者我们可以想想，其中是否蕴含着位置的金矿等待挖掘？

分享嘉宾四：陈乾龙 / 京东微信手 Q 业务部

负责微信、手Q抢购后台，以及 OpenResty 入口流量的网关

分享主题：《 基于 OpenResty 流量的防刷及容灾 》

本次分享主要介绍基于 OpenResty 搭建流量网关的实践以及开发过程中碰到的问题及优化。

六、现场礼品

因为总是看到很多同学在说elasticsearch性能不够好，集群不够稳定，询问关于elasticsearch的调优，但是每次都是一个个点的单独讲，很多时候都是case by case的解答，今天简单梳理下日常的elasticsearch使用调优，以下仅为自己日常经验之谈，如有疏漏，还请大家帮忙指正。

一、配置文件调优

elasticsearch.yml

内存锁定

bootstrap.memory_lock：true 允许 JVM 锁住内存，禁止操作系统交换出去。

zen.discovery

Elasticsearch 默认被配置为使用单播发现，以防止节点无意中加入集群。组播发现应该永远不被使用在生产环境了，否则你得到的结果就是一个节点意外的加入到了你的生产环境，仅仅是因为他们收到了一个错误的组播信号。 ES是一个P2P类型的分布式系统，使用gossip协议，集群的任意请求都可以发送到集群的任一节点，然后es内部会找到需要转发的节点，并且与之进行通信。 在es1.x的版本，es默认是开启组播，启动es之后，可以快速将局域网内集群名称，默认端口的相同实例加入到一个大的集群，后续再es2.x之后，都调整成了单播，避免安全问题和网络风暴； 单播discovery.zen.ping.unicast.hosts，建议写入集群内所有的节点及端口，如果新实例加入集群，新实例只需要写入当前集群的实例，即可自动加入到当前集群，之后再处理原实例的配置即可，新实例加入集群，不需要重启原有实例； 节点zen相关配置： discovery.zen.ping_timeout：判断master选举过程中，发现其他node存活的超时设置，主要影响选举的耗时，参数仅在加入或者选举 master 主节点的时候才起作用 discovery.zen.join_timeout：节点确定加入到集群中，向主节点发送加入请求的超时时间，默认为3s discovery.zen.minimum_master_nodes：参与master选举的最小节点数，当集群能够被选为master的节点数量小于最小数量时，集群将无法正常选举。

故障检测（ fault detection ）

两种情况下回进行故障检测，第一种是由master向集群的所有其他节点发起ping，验证节点是否处于活动状态；第二种是：集群每个节点向master发起ping，判断master是否存活，是否需要发起选举。 故障检测需要配置以下设置使用 形如： discovery.zen.fd.ping_interval 节点被ping的频率，默认为1s。 discovery.zen.fd.ping_timeout 等待ping响应的时间，默认为 30s，运行的集群中，master 检测所有节点，以及节点检测 master 是否正常。 discovery.zen.fd.ping_retries ping失败/超时多少导致节点被视为失败，默认为3。

https://www.elastic.co/guide/en/elasticsearch/reference/6.x/modules-discovery-zen.html

队列数量

不建议盲目加大es的队列数量，如果是偶发的因为数据突增，导致队列阻塞，加大队列size可以使用内存来缓存数据，如果是持续性的数据阻塞在队列，加大队列size除了加大内存占用，并不能有效提高数据写入速率，反而可能加大es宕机时候，在内存中可能丢失的上数据量。 哪些情况下，加大队列size呢？GET /_cat/thread_pool，观察api中返回的queue和rejected，如果确实存在队列拒绝或者是持续的queue，可以酌情调整队列size。

https://www.elastic.co/guide/en/elasticsearch/reference/6.x/modules-threadpool.html

内存使用

设置indices的内存熔断相关参数，根据实际情况进行调整，防止写入或查询压力过高导致OOM， indices.breaker.total.limit: 50%，集群级别的断路器，默认为jvm堆的70%； indices.breaker.request.limit: 10%，单个request的断路器限制，默认为jvm堆的60%； indices.breaker.fielddata.limit: 10%，fielddata breaker限制，默认为jvm堆的60%。

https://www.elastic.co/guide/en/elasticsearch/reference/6.x/circuit-breaker.html

根据实际情况调整查询占用cache，避免查询cache占用过多的jvm内存，参数为静态的，需要在每个数据节点配置。 indices.queries.cache.size: 5%，控制过滤器缓存的内存大小，默认为10%。接受百分比值，5%或者精确值，例如512mb。

https://www.elastic.co/guide/en/elasticsearch/reference/6.x/query-cache.html

创建shard

如果集群规模较大，可以阻止新建shard时扫描集群内全部shard的元数据，提升shard分配速度。 cluster.routing.allocation.disk.include_relocations: false，默认为true。

https://www.elastic.co/guide/en/elasticsearch/reference/6.x/disk-allocator.html

二、系统层面调优

jdk版本

当前根据官方建议，选择匹配的jdk版本；

jdk内存配置

首先，-Xms和-Xmx设置为相同的值，避免在运行过程中再进行内存分配，同时，如果系统内存小于64G，建议设置略小于机器内存的一半，剩余留给系统使用。 同时，jvm heap建议不要超过32G（不同jdk版本具体的值会略有不同），否则jvm会因为内存指针压缩导致内存浪费，详见： https://www.elastic.co/guide/cn/elasticsearch/guide/current/heap-sizing.html

交换分区

关闭交换分区，防止内存发生交换导致性能下降（部分情况下，宁死勿慢） swapoff -a

文件句柄

Lucene 使用了 大量的 文件。 同时，Elasticsearch 在节点和 HTTP 客户端之间进行通信也使用了大量的套接字，所有这一切都需要足够的文件描述符，默认情况下，linux默认运行单个进程打开1024个文件句柄，这显然是不够的，故需要加大文件句柄数 ulimit -n 65536

https://www.elastic.co/guide/en/elasticsearch/reference/6.5/setting-system-settings.html

mmap

Elasticsearch 对各种文件混合使用了 NioFs（ 注：非阻塞文件系统）和 MMapFs （ 注：内存映射文件系统）。请确保你配置的最大映射数量，以便有足够的虚拟内存可用于 mmapped 文件。这可以暂时设置： sysctl -w vm.max_map_count=262144 或者你可以在 /etc/sysctl.conf 通过修改 vm.max_map_count 永久设置它。

https://www.elastic.co/guide/cn/elasticsearch/guide/current/_file_descriptors_and_mmap.html

磁盘

如果你正在使用 SSDs，确保你的系统 I/O 调度程序是配置正确的。 当你向硬盘写数据，I/O 调度程序决定何时把数据实际发送到硬盘。 大多数默认 nix 发行版下的调度程序都叫做 cfq（完全公平队列）。但它是为旋转介质优化的： 机械硬盘的固有特性意味着它写入数据到基于物理布局的硬盘会更高效。 这对 SSD 来说是低效的，尽管这里没有涉及到机械硬盘。但是，deadline 或者 noop 应该被使用。deadline 调度程序基于写入等待时间进行优化， noop 只是一个简单的 FIFO 队列。 echo noop > /sys/block/sd/queue/scheduler

磁盘挂载

mount -o noatime,data=writeback,barrier=0,nobh /dev/sd* /esdata* 其中，noatime，禁止记录访问时间戳；data=writeback，不记录journal；barrier=0，因为关闭了journal，所以同步关闭barrier； nobh，关闭buffer_head，防止内核影响数据IO

磁盘其他注意事项

使用 RAID 0。条带化 RAID 会提高磁盘I/O，代价显然就是当一块硬盘故障时整个就故障了，不要使用镜像或者奇偶校验 RAID 因为副本已经提供了这个功能。 另外，使用多块硬盘，并允许 Elasticsearch 通过多个 path.data 目录配置把数据条带化分配到它们上面。 不要使用远程挂载的存储，比如 NFS 或者 SMB/CIFS。这个引入的延迟对性能来说完全是背道而驰的。

三、elasticsearch使用方式调优

当elasticsearch本身的配置没有明显的问题之后，发现es使用还是非常慢，这个时候，就需要我们去定位es本身的问题了，首先祭出定位问题的第一个命令：

hot_threads

GET /_nodes/hot_threads&interval=30s

抓取30s的节点上占用资源的热线程，并通过排查占用资源最多的TOP线程来判断对应的资源消耗是否正常，一般情况下，bulk，search类的线程占用资源都可能是业务造成的，但是如果是merge线程占用了大量的资源，就应该考虑是不是创建index或者刷磁盘间隔太小，批量写入size太小造成的。

https://www.elastic.co/guide/en/elasticsearch/reference/6.x/cluster-nodes-hot-threads.html

pending_tasks

GET /_cluster/pending_tasks

有一些任务只能由主节点去处理，比如创建一个新的 索引或者在集群中移动分片，由于一个集群中只能有一个主节点，所以只有这一master节点可以处理集群级别的元数据变动。在99.9999%的时间里，这不会有什么问题，元数据变动的队列基本上保持为零。在一些罕见的集群里，元数据变动的次数比主节点能处理的还快，这会导致等待中的操作会累积成队列。这个时候可以通过pending_tasks api分析当前什么操作阻塞了es的队列，比如，集群异常时，会有大量的shard在recovery，如果集群在大量创建新字段，会出现大量的put_mappings的操作，所以正常情况下，需要禁用动态mapping。

https://www.elastic.co/guide/en/elasticsearch/reference/current/cluster-pending.html

字段存储

当前es主要有doc_values，fielddata，storefield三种类型，大部分情况下，并不需要三种类型都存储，可根据实际场景进行调整： 当前用得最多的就是doc_values，列存储，对于不需要进行分词的字段，都可以开启doc_values来进行存储（且只保留keyword字段），节约内存，当然，开启doc_values会对查询性能有一定的影响，但是，这个性能损耗是比较小的，而且是值得的；

fielddata构建和管理 100% 在内存中，常驻于 JVM 内存堆，所以可用于快速查询，但是这也意味着它本质上是不可扩展的，有很多边缘情况下要提防，如果对于字段没有分析需求，可以关闭fielddata；

storefield主要用于_source字段，默认情况下，数据在写入es的时候，es会将doc数据存储为_source字段，查询时可以通过_source字段快速获取doc的原始结构，如果没有update，reindex等需求，可以将_source字段disable；

_all，ES在6.x以前的版本，默认将写入的字段拼接成一个大的字符串，并对该字段进行分词，用于支持整个doc的全文检索，在知道doc字段名称的情况下，建议关闭掉该字段，节约存储空间，也避免不带字段key的全文检索；

norms：搜索时进行评分，日志场景一般不需要评分，建议关闭；

tranlog

Elasticsearch 2.0之后为了保证不丢数据，每次 index、bulk、delete、update 完成的时候，一定触发刷新 translog 到磁盘上，才给请求返回 200 OK。这个改变在提高数据安全性的同时当然也降低了一点性能。 如果你不在意这点可能性，还是希望性能优先，可以在 index template 里设置如下参数：

{
    "index.translog.durability": "async"
}

index.translog.sync_interval：对于一些大容量的偶尔丢失几秒数据问题也并不严重的集群，使用异步的 fsync 还是比较有益的。比如，写入的数据被缓存到内存中，再每5秒执行一次 fsync ，默认为5s。小于的值100ms是不允许的。 index.translog.flush_threshold_size：translog存储尚未安全保存在Lucene中的所有操作。虽然这些操作可用于读取，但如果要关闭并且必须恢复，则需要重新编制索引。此设置控制这些操作的最大总大小，以防止恢复时间过长。达到设置的最大size后，将发生刷新，生成新的Lucene提交点，默认为512mb。

refresh_interval

执行刷新操作的频率，这会使索引的最近更改对搜索可见，默认为1s，可以设置-1为禁用刷新，对于写入速率要求较高的场景，可以适当的加大对应的时长，减小磁盘io和segment的生成；

禁止动态mapping

动态mapping的坏处： 1.造成集群元数据一直变更，导致集群不稳定； 2.可能造成数据类型与实际类型不一致； 3.对于一些异常字段或者是扫描类的字段，也会频繁的修改mapping，导致业务不可控。 动态mapping配置的可选值及含义如下： true：支持动态扩展，新增数据有新的字段属性时，自动添加对于的mapping，数据写入成功 false：不支持动态扩展，新增数据有新的字段属性时，直接忽略，数据写入成功 strict：不支持动态扩展，新增数据有新的字段时，报错，数据写入失败

批量写入

批量请求显然会大大提升写入速率，且这个速率是可以量化的，官方建议每次批量的数据物理字节数5-15MB是一个比较不错的起点，注意这里说的是物理字节数大小。文档计数对批量大小来说不是一个好指标。比如说，如果你每次批量索引 1000 个文档，记住下面的事实： 1000 个 1 KB 大小的文档加起来是 1 MB 大。 1000 个 100 KB 大小的文档加起来是 100 MB 大。 这可是完完全全不一样的批量大小了。批量请求需要在协调节点上加载进内存，所以批量请求的物理大小比文档计数重要得多。 从 5–15 MB 开始测试批量请求大小，缓慢增加这个数字，直到你看不到性能提升为止。然后开始增加你的批量写入的并发度（多线程等等办法）。 用iostat 、 top 和 ps 等工具监控你的节点，观察资源什么时候达到瓶颈。如果你开始收到 EsRejectedExecutionException ，你的集群没办法再继续了：至少有一种资源到瓶颈了。或者减少并发数，或者提供更多的受限资源（比如从机械磁盘换成 SSD），或者添加更多节点。

索引和shard

es的索引，shard都会有对应的元数据，且因为es的元数据都是保存在master节点，且元数据的更新是要hang住集群向所有节点同步的，当es的新建字段或者新建索引的时候，都会要获取集群元数据，并对元数据进行变更及同步，此时会影响集群的响应，所以需要关注集群的index和shard数量，建议如下： 1.使用shrink和rollover api，相对生成合适的数据shard数； 2.根据数据量级及对应的性能需求，选择创建index的名称，形如：按月生成索引：test-YYYYMM，按天生成索引：test-YYYYMMDD； 3.控制单个shard的size，正常情况下，日志场景，建议单个shard不大于50GB，线上业务场景，建议单个shard不超过20GB；

segment merge

段合并的计算量庞大， 而且还要吃掉大量磁盘 I/O。合并在后台定期操作，因为他们可能要很长时间才能完成，尤其是比较大的段。这个通常来说都没问题，因为大规模段合并的概率是很小的。 如果发现merge占用了大量的资源，可以设置： index.merge.scheduler.max_thread_count: 1 特别是机械磁盘在并发 I/O 支持方面比较差，所以我们需要降低每个索引并发访问磁盘的线程数。这个设置允许 max_thread_count + 2 个线程同时进行磁盘操作，也就是设置为 1 允许三个线程。 对于 SSD，你可以忽略这个设置，默认是 Math.min(3, Runtime.getRuntime().availableProcessors() / 2) ，对 SSD 来说运行的很好。 业务低峰期通过force_merge强制合并segment，降低segment的数量，减小内存消耗； 关闭冷索引，业务需要的时候再进行开启，如果一直不使用的索引，可以定期删除，或者备份到hadoop集群；

自动生成_id

当写入端使用特定的id将数据写入es时，es会去检查对应的index下是否存在相同的id，这个操作会随着文档数量的增加而消耗越来越大，所以如果业务上没有强需求，建议使用es自动生成的id，加快写入速率。

routing

对于数据量较大的业务查询场景，es侧一般会创建多个shard，并将shard分配到集群中的多个实例来分摊压力，正常情况下，一个查询会遍历查询所有的shard，然后将查询到的结果进行merge之后，再返回给查询端。此时，写入的时候设置routing，可以避免每次查询都遍历全量shard，而是查询的时候也指定对应的routingkey，这种情况下，es会只去查询对应的shard，可以大幅度降低合并数据和调度全量shard的开销。

使用alias

生产提供服务的索引，切记使用别名提供服务，而不是直接暴露索引名称，避免后续因为业务变更或者索引数据需要reindex等情况造成业务中断。

避免宽表

在索引中定义太多字段是一种可能导致映射爆炸的情况，这可能导致内存不足错误和难以恢复的情况，这个问题可能比预期更常见，index.mapping.total_fields.limit ，默认值是1000

避免稀疏索引

因为索引稀疏之后，对应的相邻文档id的delta值会很大，lucene基于文档id做delta编码压缩导致压缩率降低，从而导致索引文件增大，同时，es的keyword，数组类型采用doc_values结构，每个文档都会占用一定的空间，即使字段是空值，所以稀疏索引会造成磁盘size增大，导致查询和写入效率降低。

1.滴滴Elasticsearch多集群架构实践。 http://t.cn/EUNLkNU

2. 用es作为存储进行机器学习的python库。 http://t.cn/EUWeDoI

3. 一周热点：职场寒冬，给你讲四个小故事 http://t.cn/Eydyut9

• 编辑：bsll

• 归档：https://elasticsearch.cn/article/6200

• 订阅：https://tinyletter.com/elastic-daily

介绍

本次想和大家分享一款Elasticsearch分词插件，该插件是基于天津海量信息股份有限公司的中文分词核心开发的。海量分词针对大数据检索场景专门做了定制和优化，更贴近搜索需求，整体分词的性能也是非常高效。

本文章有广告成分。但希望将公司研究成果分享出来，给大家实际工作中多一种选择...

海量分词检索优化点

• 地名方面海量分词5.0可以识别并检索出关于地名后缀的结果

  可以通过搜索“河南”得到“河南省”的结果，搜索“天津”得到“天津市”的搜索结果，而不是简单河南、天津的识别。

• 著名人物的人名识别更精准，如刘翔、傅莹等

  部分分词器处理中文分词只有两种方式：一种是单字（unigrams）形式，即简单粗暴的将中文的每一个汉字作为一个词（token）分开；另一种是两字（bigrams）的，也就是任意相邻的两个汉字作为一个词分开。这种简单粗暴的切分方式无法实现时效性较新的人名识别，如刘翔、傅莹等会被识别为单字切开。

• 外国人名识别方面海量可以将人名识别智能识别

  “玛利亚 凯利”、“乔治·史密斯”、“玛丽·戴维斯”将完整的外国人名识别出姓氏和名，如“乔治·史密斯”可以被识别为“乔治”和 “史密斯”。

• 常见词的品牌名称识别方面，海量分词5.0识别的结果中包含实际意义的品牌名称

  如“乐高”，“吉米作为简单的词，可以被识别，但是词放在文档语境中有其品牌的属性，海量分词识别的结果中可以准确搜索出品牌的结果。

• 机构名识别方面

  海量分词5.0可以识别完整的机构名称，如“天津海量信息技术股份有限公司”，可以完整的识别出全称。

海量分词性能评测

评测用例

本次评测选取的语料一共三个。一个是2MB的海量测试语料，一个是4MB的北大语料（新版旧版各2MB），一个是9.4GB海量的线上实际数据

评测指标

本次评测是在开源评测程序上修改而来，评测指标有分词速度、行数完美率、字数完美率（该指标仅供参考）、内存消耗

评测结果

2MB海量测试语料

可以看到海量分词的行数完美率是最高的，而且速度十分优异；仅有的一个比海量分词速度快的算法是一个追求极限性能舍弃准确率的算法

4MB北大语料

（注：THULAC和stanford由于速度问题，不纳入评测）

可以看到海量的速度和行数完美率都很优异而且达到了兼顾，行数完美率只落后更高的ltp算法1.4个百分点，速度却是它的三十多倍

9.4GB线上数据

（注：THULAC和stanford由于速度问题，不纳入评测）

本表格中分词顺序按（4MB北大语料的）行数完美率进行排序，越靠前的（4MB北大语料的）行数完美率越高

可以看出海量的分词速度十分优秀，分词速度拉开了大多数分词数倍，相比于行数完美率小幅领先的ltp要快几十倍

海量分词插件使用方法

安装使用

• 下载安装 - 地址: https://github.com/HylandaOpen/elasticsearch-analysis-hlseg/releases

  unzip plugin to folder `your-es-root/plugins/`

• 使用 elasticsearch-plugin 安装

  ./bin/elasticsearch-plugin install https://github.com/HylandaOpen/elasticsearch-analysis-hlseg/releases/download/v6.4.2/elasticsearch-analysis-hlseg-6.4.2.zip

• 重启es集群

实例（借用github-ik分词插件的实例）

1.创建index

curl -XPUT http://localhost:9200/hylanda_seg

2.配置mapping

curl -XPOST http://localhost:9200/hylanda_seg/data/_mapping -H 'Content-Type:application/json' -d'
{
  "properties": {
    "msg": {
      "type": "text",
      "analyzer": "hlseg_search"
    }
  }
}'

3.插入测试数据

curl -XPOST http://localhost:9200/hylanda_seg/data/1 -H 'Content-Type:application/json' -d'
{"content":"美国留给伊拉克的是个烂摊子吗"}
'

curl -XPOST http://localhost:9200/hylanda_seg/data/2 -H 'Content-Type:application/json' -d'
{"content":"公安部：各地校车将享最高路权"}
'

curl -XPOST http://localhost:9200/hylanda_seg/data/3 -H 'Content-Type:application/json' -d'
{"content":"中韩渔警冲突调查：韩警平均每天扣1艘中国渔船"}
'

curl -XPOST http://localhost:9200/hylanda_seg/data/4 -H 'Content-Type:application/json' -d'
{"content":"中国驻洛杉矶领事馆遭亚裔男子枪击 嫌犯已自首"}
'

4.查询

curl -XPOST http://localhost:9200/hylanda_seg/data/_search  -H 'Content-Type:application/json' -d'
{
  "query": {
    "match": {
      "content": "中国"
    }
  },
  "highlight": {
    "fields": {
      "content": {}
    }
  }
}
'

返回结果

{
  "took" : 11,
  "timed_out" : false,
  "_shards" : {
    "total" : 5,
    "successful" : 5,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : 2,
    "max_score" : 0.5754429,
    "hits" : [
      {
        "_index" : "hylanda_seg",
        "_type" : "data",
        "_id" : "4",
        "_score" : 0.5754429,
        "_source" : {
          "content" : "中韩渔警冲突调查：韩警平均每天扣1艘中国渔船"
        },
        "highlight" : {
          "content" : [
            "中韩渔警冲突调查：韩警平均每天扣1艘<em>中国</em>渔船"
          ]
        }
      },
      {
        "_index" : "hylanda_seg",
        "_type" : "data",
        "_id" : "5",
        "_score" : 0.2876821,
        "_source" : {
          "content" : "中国驻洛杉矶领事馆遭亚裔男子枪击 嫌犯已自首"
        },
        "highlight" : {
          "content" : [
            "<em>中国</em>驻洛杉矶领事馆遭亚裔男子枪击 嫌犯已自首"
          ]
        }
      }
    ]
  }
}

字典配置

海量分词分为基础词词典CoreDict.dat和自定义词典userDict_utf8.txt。基础词词典在dictionary目录下，需要将CoreDict.zip解压后放在config目录下，可以通过修改config下的userDict_utf8.txt来更新自定义词典

自定义词典格式如下

1.用户自定义词典采用文本格式，utf-8编码，每行一个词

2.每个词包含三列属性，分别是词串、词的属性以及idf值的加权等级，并以Tab作为分隔，其中除了词串必填外，其他列可以不填，不填写则系统采用默认值

3.“#”表示注释，会在加载时被忽略

4.词的属性以西文逗号分隔，可以是词性、停止词标志或者自定义属性

5.词性标记参考北大标准，用于词性标注时参考，该项不填则默认为名词

6.停止词标志为：stopword，由SegOption.outputStopWord来控制是否输出停止词

7.自定义属性不参与分词过程，分词结果中若Token.userTag不为空，则可以获取到该词的自定义属性。

8.idf值的加权分5级，从低到高的定义是idf-lv1 — idf-lv5，等级越高则该词在关键词计算时的权重会越大，若不填写该值则系统默认是idf-lv3(中等权重）

	BasicDataSource basicDataSource = new BasicDataSource();

	// 创建连接池 一次性创建多个连接池



	// 连接池 创建连接 ---需要四个参数

	basicDataSource.setDriverClassName("com.elasticsearch.jdbc.ElasticsearchDriver");

	basicDataSource.setUrl("jdbc:elasticsearch://127.0.0.1:5000");



	// 从连接池中获取连接

	Connection conn = basicDataSource.getConnection();

	String sql = "select SRC_IP,SRC_PORT from \"my_test-*\"";

	Statement stmt = conn.createStatement();

	ResultSet rs = stmt.executeQuery(sql);

	while (rs.next()) {

		System.out.println(rs.getString("SRC_IP"));

	}

	basicDataSource.close();





	String sql = "select SRC_IP,SRC_PORT from my_test* where SRC_PORT between 10 and 100 limit 1000";

	String url = "jdbc:elasticsearch://127.0.0.1:5000";

	Connection connection = DriverManager.getConnection(url, "test", null);

	Statement statement = connection.createStatement();

	ResultSet rs = statement.executeQuery(sql);

	ResultSetMetaData meta = rs.getMetaData();

	String columns = "|";

	for (int i = 0; i < meta.getColumnCount(); i++) {

		columns += meta.getColumnLabel(i) + " | ";

	}

	System.out.println(columns);

	while (rs.next()) {

		String row = "|";

		for (int i = 0; i < meta.getColumnCount(); i++) {

			row += rs.getString(i) + " | ";

		}

		System.out.println(row);

	}