1.如何在Elasticsearch中查找和删除重复文档
http://t.cn/EUxkESo
2.Elasticsearch线程池分析
http://t.cn/EUxkDNg
3.Elasticsearch Pipeline Aggregations指南
http://t.cn/EUxFzZX

编辑：金桥
归档：https://elasticsearch.cn/article/6195
订阅：https://tinyletter.com/elastic-daily

ES-Hadoop打通Elasticsearch和Hadoop

介绍

Elasticsearch作为强大的搜索引擎，Hadoop HDFS是分布式文件系统。

ES-Hadoop是一个深度集成Hadoop和ElasticSearch的项目，也是ES官方来维护的一个子项目。Elasticsearch可以将自身的Document导入到HDFS中用作备份；同时也可以将存储在HDFS上的结构化文件导入为ES中的Document，通过实现Hadoop和ES之间的输入输出，可以在Hadoop里面对ES集群的数据进行读取和写入，充分发挥Map-Reduce并行处理的优势，为Hadoop数据带来实时搜索的可能。

ES-Hadoop插件支持Map-Reduce、Cascading、Hive、Pig、Spark、Storm、yarn等组件。

ES-Hadoop整个数据流转图如下：

环境配置

• Elasticsearch 5.0.2
  
• Centos 7
  
• elasticsearch-hadoop 5.0.2
  
• repository-hdfs-5.0.2
  
  

  Elasticsearch备份数据到HDFS
  

  
  

  介绍
  

  
  Elasticsearch副本提供了数据高可靠性，在部分节点丢失的情况下不中断服务；但是副本并不提供对灾难性故障的保护，同时在运维人员误操作情况下也不能保障数据的可恢复性。对于这种情况，我们需要对Elasticsearch集群数据的真正备份。
  
  通过快照的方式，将Elasticsearch集群中的数据备份到HDFS上，这样数据既存在于Elasticsearch集群中，有存在于HDFS上。当ES集群出现不可恢复的故障时，可以将数据从HDFS上快速恢复。
  
  

  操作步骤
  

  
  

• 下载插件 <https://artifacts.elastic.co/d ... gt%3B 保存在/usr/local下
  
  
• 安装插件
  
  <br /> cd /usr/local/es/elasticsearch-5.0.2/bin<br /> ./elasticsearch-plugin install file:///usr/local/repository-hdfs-5.0.2.zip<br />
  
  
• 安装成功后需要重启Elasticsearch
  
  

  备份与恢复
  

  
  

• 构建一个仓库
  
  <br /> PUT <a href="http://192.168.10.74:9200/_snapshot/backup" rel="nofollow" target="_blank">http://192.168.10.74:9200/_snapshot/backup</a><br /> { <br /> "type": "hdfs", <br /> "settings": { <br /> "uri": "hdfs://192.168.10.170:9000", <br /> "path": "/es", <br /> "conf_location": "/usr/local/hadoop/etc/hadoop/hdfs-site.xml" <br /> }<br /> }<br />
  
  
• 备份快照
  
  <br /> PUT <a href="http://192.168.10.74:9200/_snapshot/backup/snapshot_users?wait_for_completion=true" rel="nofollow" target="_blank">http://192.168.10.74:9200/_sna ... Dtrue</a><br /> {<br /> "indices": "users", //备份users的index，注意不设置这个属性，默认是备份所有index<br /> "ignore_unavailable": true,<br /> "include_global_state": false<br /> }<br />
  
  
• 恢复快照
  
  <br /> POST <a href="http://192.168.10.74:9200/_snapshot/backup/snapshot_users/_restore" rel="nofollow" target="_blank">http://192.168.10.74:9200/_sna ... store</a><br /> {<br /> "indices": "users", //指定索引恢复，不指定就是所有<br /> "ignore_unavailable": true, //忽略恢复时异常索引<br /> "include_global_state": false //是否存储全局转态信息,fasle代表有一个或几个失败，不会导致整个任务失败<br /> }<br />
  
  
  

  整合Spark与Elasticsearch
  

  
  

  整体思路
  

  
  

• 数据首先存储在HDFS上，可以通过Spark SQL直接导入到ES中
  
• Spark SQL可以直接通过建立Dataframe或者临时表连接ES，达到搜索优化、减少数据量和数据筛选的目的，此时数据只在ES内存中而不再Spark SQL中
  
• 筛选后的数据重新导入到Spark SQL中进行查询
  
  

  引入依赖
  

  
  ```java
  

  
  org.elasticsearch
  elasticsearch-hadoop
  5.0.2
  
  ```
  
  
  
  ### 具体流程
  
  
• 数据在HDFS上，数据存储在HDFS的每个DataNode的block上
  
  
  
  

  
  
  
  

• 数据加载到Spark SQL
  
  
  • 数据从HDFS加载到Spark SQL中，以RDD形式存储
    
    java<br /> JavaRDD<String> textFile = spark.read().textFile("hdfs://192.168.10.170:9000/csv/user.csv")<br />
    
    
  • 添加数据结构信息转换为新的RDD
    
    java<br /> JavaRDD<UserItem> dataSplits = textFile.map(line -> {<br /> String records = line.toString().trim();<br /> String record = records.substring(0,records.length() - 1).trim();<br /> String[] parts = record.split("\\|");<br /> UserItem u = new UserItem();<br /> u.setName(parts[0]);<br /> u.setAge(parts[1]);<br /> u.setHeight(parts[2]);<br /> return u;<br /> });<br />
    
    
  • 根据新的RDD创建DataFrame
    
    java<br /> DataSet<Row> ds = spark.createDataFrame(dataSplits, UserItem.class);<br />
    
    
    
    

    
    
    
    

  • 由Dataset创建索引，并写入ES
    
    <br /> JavaEsSparkSQL.saveToEs(ds, "es_spark/users");<br />
    
    
• 数据在ES中建立索引
  
  
  
  

  
  
  
  

• Spark SQL通过索引对ES中的数据进行查询
  
  java<br /> SparkSession spark = SparkSession.builder().appName("es-spark").master("local").config("es.index.auto.create", true).getOrCreate();<br /> Map<String, String> options = new HashMap<>();<br /> options.put("pushdown", "true");<br /> options.put("es.nodes","192.168.10.74:9200");<br /> <br /> Dataset<Row> df = spark.read().options(options).format("org.elasticsearch.spark.sql").load("es_spark/users");<br /> df.createOrReplaceTempView("users");<br /> <br /> Dataset<Row> userSet = spark.sql("SELECT name FORM users WHERE age >=10 AND age <= 20");<br /> userSet.show();<br />
  
  

  结束
  

  
  ES-Hadoop无缝打通了ES和Hadoop两个非常优秀的框架，从而让ES的强大检索性能帮助我们快速分析海量数据。

1. 快来学习下如何便捷地记录文档插入 es 的时间？
http://t.cn/EUfESt7
2. 6.5.3发布了，还没升级到6的同学又多了一个新的选择！
http://t.cn/EUfEHvA
3. (自备梯子)如何不停机修改索引的 mapping？
http://t.cn/EUfErLr

编辑：rockybean
归档：https://elasticsearch.cn/article/6193
订阅：https://tinyletter.com/elastic-daily

为什么我的数据中没有updateTime 和 createTime 字段的；  理论上是不会执行if 里面的代码才对的；  但是为什么看日志输出好像是执行了if代码块的代码呢 
 
下面的是数据源， 并没有time字段的

{

  "仓ku": "华南",

   "originName":  "",

   "Code":  "23248",

   "BrandName":  "",

   "originCode":  null,

   "CategoryName":  "原厂"

}

1. 背景

Elasticsearch可广泛应用于日志分析、全文检索、结构化数据分析等多种场景，大幅度降低维护多套专用系统的成本，在开源社区非常受欢迎。然而Elasticsearch为满足多种不同的使用场景，底层组合使用了多种数据结构，部分数据结构对具体的用户使用场景可能是冗余的，从而导致默认情况下无法达到性能和成本最优化。
幸运的是，Elasticsearch提供非常灵活的模板配置能力，用户可以按需进行优化。多数情况下，用户结合使用场景进行优化后，Elasticsearch的性能都会有数倍的提升，成本也对应有倍数级别的下降。本文主要介绍不同日志使用场景下的调优经验。

2. 日志处理基本流程

日志处理的基本流程包含：日志采集 -> 数据清洗 -> 存储 -> 可视化分析。Elastic Stack提供完整的日志解决方案，帮助用户完成对日志处理全链路的管理，推荐大家使用。每个流程的处理如下：

• 日志采集：从业务所在的机器上，较实时的采集日志传递给下游。常用开源组件如Beats、Logstash、Fluentd等。
  
• 数据清洗：利用正则解析等机制，完成日志从文本数据到结构化数据的转换。用户可使用Logstash 或 Elasticsearch Ingest模块等完成数据清洗。
  
• 存储：使用Elasticsearch对数据进行持久存储，并提供全文搜索和分析能力。
  
• 可视化分析：通过图形界面，完成对日志的搜索分析，常用的开源组件如Kibana、Grafana。
  
  
  使用Elastic Stack处理日志的详细过程，用户可参考官方文章[Getting started with the Elastic Stack](https://www.elastic.co/guide/e ... k.html)，这里不展开介绍。
  
  
  

  3. 日志场景调优
  

         对于Elasticsearch的通用调优，之前分享的文章[Elasticsearch调优实践](https://cloud.tencent.com/developer/article/1156231)，详细介绍了Elasticsearch在性能、稳定性方面的调优经验。而对于日志场景，不同的场景使用方式差别较大，这里主要介绍常见使用方式下，性能和成本的优化思路。
  
  

  3.1 基础场景
  

  对于多数简单日志使用场景，用户一般只要求存储原始日志，并提供按关键字搜索日志记录的能力。对于此类场景，用户可跳过数据清洗阶段，并参考如下方式进行优化：
  
  

• 建议打开最优压缩，一般可降低40%存储。
  
• 设置原始日志字段（message）为text，去除keyword类型子字段，提供全文搜索能力，降低存储。
  
• 关闭_all索引，前面已通过message提供全文搜索能力。
  
• 对于其他字符串字段，统一设置为keyword类型，避免默认情况下字符串字段同时存储text、keyword两种类型的数据。
  
• 使用开源组件（如Beats）上报数据时会包含较多辅助信息，用户可通过修改组件配置文件进行裁剪。
  
  这样去除message的keyword子字段、_all等冗余信息后，再加上最优压缩，可以保证数据相对精简。下面给出这类场景的常用模板，供用户参考：
  <br /> {<br /> "order": 5,<br /> "template": "my_log_*",<br /> "settings": {<br /> "translog.durability": "async",<br /> "translog.sync_interval": "5s",<br /> "index.refresh_interval": "30s",<br /> "index.codec": "best_compression" # 最优压缩<br /> },<br /> "mappings": {<br /> "_default_": {<br /> "_all": { # 关闭_all索引<br /> "enabled": false<br /> },<br /> "dynamic_templates": [<br /> {<br /> "log": { # 原始日志字段，分词建立索引<br /> "match": "message",<br /> "mapping": {<br /> "type": "text"<br /> }<br /> }<br /> },<br /> {<br /> "strings": { # 其他字符串字段，统一设置为keyword类型<br /> "match_mapping_type": "string",<br /> "mapping": {<br /> "type": "keyword"<br /> }<br /> }<br /> }<br /> ]<br /> }<br /> }<br /> }<br />
  
  
  

  3.2 精准搜索场景
  

  对于部分用户，普通的全文检索并不能满足需求，希望精准搜索日志中的某部分，例如每条日志中包含程序运行时多个阶段的耗时数据，对具体一个阶段的耗时进行搜索就比较麻烦。对于此类场景，用户可基于基础场景，进行如下调整：
  
  

• 清洗过程中，可仅解析出需要精准搜索的部分作为独立字段，用于精准搜索。
  
• 对于精准搜索字段，如果无排序/聚合需求，可以关闭doc_values；对于字符串，一般使用keyword，可按需考虑使用text。
  
  下面给出这类场景的常用模板，供用户参考：
  <br /> {<br /> "order": 5,<br /> "template": "my_log_*",<br /> "settings": {<br /> "translog.durability": "async",<br /> "translog.sync_interval": "5s",<br /> "index.refresh_interval": "30s",<br /> "index.codec": "best_compression" # 最优压缩<br /> },<br /> "mappings": {<br /> "_default_": {<br /> "_all": { # 关闭_all索引<br /> "enabled": false<br /> },<br /> "dynamic_templates": [<br /> {<br /> "log": { # 原始日志字段，分词建立索引<br /> "match": "message",<br /> "mapping": {<br /> "type": "text"<br /> }<br /> }<br /> },<br /> {<br /> "precise_fieldx": { # 精准搜索字段<br /> "match": "fieldx",<br /> "mapping": {<br /> "type": "keyword",<br /> "doc_values": false<br /> }<br /> }<br /> },<br /> {<br /> "strings": { # 其他字符串字段，统一设置为keyword类型<br /> "match_mapping_type": "string",<br /> "mapping": {<br /> "type": "keyword"<br /> }<br /> }<br /> }<br /> ]<br /> }<br /> }<br /> }<br />
  
  

  3.3 统计分析场景
  

  对于某些场景，日志包含的主要是程序运行时输出的统计信息，用户通常会完全解析日志进行精确查询、统计分析，而是否保存原始日志关系不大。对于此类场景，用户可进行如下调整：
  
  

• 清洗过程中，解析出所有需要的数据作为独立字段；原始日志非必要时，建议去除。
  
• 如果有强需求保留原始日志，可以设置该字段enabled属性为false，只存储不索引。
  
• 多数字段保持默认即可，会自动建立索引、打开doc_values，可用于查询、排序、聚合。
  
• 对部分无排序/聚合需求、开销高的字段，可以关闭doc_values。
  
  
  下面给出这类场景的常用模板，供用户参考：
  
  
  <br /> {<br /> "order": 5,<br /> "template": "my_log_*",<br /> "settings": {<br /> "translog.durability": "async",<br /> "translog.sync_interval": "5s",<br /> "index.refresh_interval": "30s",<br /> "index.codec": "best_compression" # 最优压缩<br /> },<br /> "mappings": {<br /> "_default_": {<br /> "_all": { # 关闭_all索引<br /> "enabled": false<br /> },<br /> "dynamic_templates": [<br /> {<br /> "log": { # 原始日志字段，关闭索引<br /> "match": "message",<br /> "mapping": {<br /> "enabled": false<br /> }<br /> }<br /> },<br /> {<br /> "index_only_fieldx": { # 仅索引的字段，无排序/聚合需求<br /> "match": "fieldx",<br /> "mapping": {<br /> "type": "keyword",<br /> "doc_values": false<br /> }<br /> }<br /> },<br /> {<br /> "strings": { # 其他字符串字段，统一设置为keyword类型<br /> "match_mapping_type": "string",<br /> "mapping": {<br /> "type": "keyword"<br /> }<br /> }<br /> }<br /> ]<br /> }<br /> }<br /> }<br />
  

  ES 5.1及之后的版本，支持关键字查询时[自动选择目标字段](https://www.elastic.co/guide/e ... _field)，用户没有必要再使用原始日志字段提供不指定字段进行查询的能力。
  
  
  

  4. 小结
  

  日志的使用方式比较灵活，本文结合常见的客户使用方式，从整体上对性能、成本进行优化。用户也可结合自身业务场景，参考文章[Elasticsearch调优实践](https://cloud.tencent.com/developer/article/1156231)进行更细致的优化。
  
  
  
  
  
  
  

当我们搭建elasitc stack集群时，大多数时候会在我们的架构中加入kafka作为消息缓冲区，即从beats -> kafka -> logstash -> elasticsearch这样的一个消息流。使用kafka可以给我们带来很多便利，但是也让我们需要额外多维护一套组件，elasitc stack本身已经提供了monitoring的功能，我们可以方便的从kibana上监控各个组件中各节点的可用性，吞吐和性能等各种指标，但kafka作为架构中的组件之一却游离在监控之外，相当不合理。

幸而elastic真的是迭代的相当快，在metricbeat上很早就有了对kafka的监控，但一直没有一个直观的dashboard，终于在6.5版本上，上新了kafka dashboard。我们来看一下吧。

安装和配置metricbeat

[安装包下载地址](https://www.elastic.co/downloads/beats/metricbeat)，下载后，自己安装。
然后，将/etc/metricbeat/modules.d/kafka.yml.disable文件重命名为/etc/metricbeat/modules.d/kafka.yml。(即打开kafka的监控)。稍微修改一下文件内容, 注意，这里需填入所有你需要监控的kafka服务器的地址：

```

Module: kafka

Docs: https://www.elastic.co/guide/e ... .html

• module: kafka
  metricsets:
  
  • partition
    
  • consumergroup
    period: 20s
    hosts: ["10...:9092","10...:9092","10...:9092","10...:9092"]
    
    

    client_id: metricbeat
    

    retries: 3
    

    backoff: 250ms
    

    
    

    List of Topics to query metadata for. If empty, all topics will be queried.
    

    topics: []
    

    
    

    Optional SSL. By default is off.
    

    List of root certificates for HTTPS server verifications
    

    ssl.certificate_authorities: ["/etc/pki/root/ca.pem"]
    

    
    

    Certificate for SSL client authentication
    

    ssl.certificate: "/etc/pki/client/cert.pem"
    

    
    

    Client Certificate Key
    

    ssl.key: "/etc/pki/client/cert.key"
    

    
    

    SASL authentication
    

    username: ""
    

    password: ""
    

    ```
    运行metricbeat，这里，一定要注意enable kibana dashboard。
    
    然后就可以在kibana里面看到：
    
    

    

    
    

    

    
    
    
    这样，我们就可以通过sentinl等类似的插件，自动做kafka的告警等功能了

@[toc]
搭建Elasitc stack集群时，我们往往把大部分注意力放在集群的搭建，索引的优化，分片的设置上等具体的调优参数上，很少有人会去关心Elasitc stack的日志配置的问题，大概是觉得，日志应该是一个公共的问题，默认的配置应该已经为我们处理好了。但很不幸，在不同的机器配置或者不同的运营策略下，如果采用默认的配置，会给我们带来麻烦。

默认配置带来的麻烦

以下例子是默认情况下，当Elasitc stack集群运行超过3个月之后的情况：

elasticsearch

elasticsearch默认情况下会每天rolling一个文件，当到达2G的时候，才开始清除超出的部分，当一个文件只有几十K的时候，文件会一直累计下来。

logstash

一直增长的gc文件和不停增多的rolling日志文件

kibana

默认日志输出到kibana.out文件当中，这个文件会变得越来越大

kafka

这里提到kafka是因为在大部分的架构当中，我们都会用到kafka作为中间件数据缓冲区，因此不得不维护kafka集群。同样，如果不做特定的配置，也会遇到日志的问题：不停增多的rolling日志文件

原因是kafka的默认log4j配置是使用DailyRollingFileAppender每隔一个小时生成一个文件 '.'yyyy-MM-dd-HH：
<br /> log4j.appender.stdout=org.apache.log4j.ConsoleAppender<br /> log4j.appender.stdout.layout=org.apache.log4j.PatternLayout<br /> log4j.appender.stdout.layout.ConversionPattern=[%d] %p %m (%c)%n<br /> <br /> log4j.appender.kafkaAppender=org.apache.log4j.DailyRollingFileAppender<br /> log4j.appender.kafkaAppender.DatePattern='.'yyyy-MM-dd-HH<br /> log4j.appender.kafkaAppender.File=${kafka.logs.dir}/server.log<br /> log4j.appender.kafkaAppender.layout=org.apache.log4j.PatternLayout<br /> log4j.appender.kafkaAppender.layout.ConversionPattern=[%d] %p %m (%c)%n<br /> <br /> log4j.appender.stateChangeAppender=org.apache.log4j.DailyRollingFileAppender<br /> log4j.appender.stateChangeAppender.DatePattern='.'yyyy-MM-dd-HH<br /> log4j.appender.stateChangeAppender.File=${kafka.logs.dir}/state-change.log<br /> log4j.appender.stateChangeAppender.layout=org.apache.log4j.PatternLayout<br /> log4j.appender.stateChangeAppender.layout.ConversionPattern=[%d] %p %m (%c)%n<br /> <br /> log4j.appender.requestAppender=org.apache.log4j.DailyRollingFileAppender<br /> log4j.appender.requestAppender.DatePattern='.'yyyy-MM-dd-HH<br /> log4j.appender.requestAppender.File=${kafka.logs.dir}/kafka-request.log<br /> log4j.appender.requestAppender.layout=org.apache.log4j.PatternLayout<br /> log4j.appender.requestAppender.layout.ConversionPattern=[%d] %p %m (%c)%n<br /> <br /> log4j.appender.cleanerAppender=org.apache.log4j.DailyRollingFileAppender<br /> log4j.appender.cleanerAppender.DatePattern='.'yyyy-MM-dd-HH<br /> log4j.appender.cleanerAppender.File=${kafka.logs.dir}/log-cleaner.log<br /> log4j.appender.cleanerAppender.layout=org.apache.log4j.PatternLayout<br /> log4j.appender.cleanerAppender.layout.ConversionPattern=[%d] %p %m (%c)%n<br /> <br /> log4j.appender.controllerAppender=org.apache.log4j.DailyRollingFileAppender<br /> log4j.appender.controllerAppender.DatePattern='.'yyyy-MM-dd-HH<br /> log4j.appender.controllerAppender.File=${kafka.logs.dir}/controller.log<br /> log4j.appender.controllerAppender.layout=org.apache.log4j.PatternLayout<br /> log4j.appender.controllerAppender.layout.ConversionPattern=[%d] %p %m (%c)%n<br /> <br /> log4j.appender.authorizerAppender=org.apache.log4j.DailyRollingFileAppender<br /> log4j.appender.authorizerAppender.DatePattern='.'yyyy-MM-dd-HH<br /> log4j.appender.authorizerAppender.File=${kafka.logs.dir}/kafka-authorizer.log<br /> log4j.appender.authorizerAppender.layout=org.apache.log4j.PatternLayout<br /> log4j.appender.authorizerAppender.layout.ConversionPattern=[%d] %p %m (%c)%n<br />

解决方案

因此，对于我们需要维护的这几个组件，需要配置合理的日志rotate策略。一个比较常用的策略就是时间+size，每天rotate一个日志文件或者每当日志文件大小超过256M，rotate一个新的日志文件，并且最多保留7天之内的日志文件。

elasticsearch 

通过修改log4j2.properties文件来解决。该文件在/etc/elasticsesarch目录下（或者config目录）。
默认配置是：
<br /> appender.rolling.type = RollingFile <br /> appender.rolling.name = rolling<br /> appender.rolling.fileName = ${sys:es.logs.base_path}${sys:file.separator}${sys:es.logs.cluster_name}.log <br /> appender.rolling.layout.type = PatternLayout<br /> appender.rolling.layout.pattern = [%d{ISO8601}][%-5p][%-25c{1.}] [%node_name]%marker %.-10000m%n<br /> appender.rolling.filePattern = ${sys:es.logs.base_path}${sys:file.separator}${sys:es.logs.cluster_name}-%d{yyyy-MM-dd}-%i.log.gz <br /> appender.rolling.policies.type = Policies<br /> appender.rolling.policies.time.type = TimeBasedTriggeringPolicy <br /> appender.rolling.policies.time.interval = 1 <br /> appender.rolling.policies.time.modulate = true <br /> appender.rolling.policies.size.type = SizeBasedTriggeringPolicy <br /> appender.rolling.policies.size.size = 256MB <br /> appender.rolling.strategy.type = DefaultRolloverStrategy<br /> appender.rolling.strategy.fileIndex = nomax<br /> appender.rolling.strategy.action.type = Delete <br /> appender.rolling.strategy.action.basepath = ${sys:es.logs.base_path}<br /> appender.rolling.strategy.action.condition.type = IfFileName <br /> appender.rolling.strategy.action.condition.glob = ${sys:es.logs.cluster_name}-* <br /> appender.rolling.strategy.action.condition.nested_condition.type = IfAccumulatedFileSize <br /> appender.rolling.strategy.action.condition.nested_condition.exceeds = 2GB <br />
以上默认配置，会保存2GB的日志，只有累计的日志大小超过2GB的时候，才会删除旧的日志文件。
建议改为如下配置，仅保留最近7天的日志
<br /> appender.rolling.strategy.type = DefaultRolloverStrategy<br /> appender.rolling.strategy.action.type = Delete<br /> appender.rolling.strategy.action.basepath = ${sys:es.logs.base_path}<br /> appender.rolling.strategy.action.condition.type = IfFileName<br /> appender.rolling.strategy.action.condition.glob = ${sys:es.logs.cluster_name}-*<br /> appender.rolling.strategy.action.condition.nested_condition.type = IfLastModified<br /> appender.rolling.strategy.action.condition.nested_condition.age = 7D<br />
这里必须注意，log4j2会因为末尾的空格导致无法识别配置

logstash

与elasticsearch类似，通过修改log4j2.properties文件来解决。该文件在/etc/logstash目录下（或者config目录）。
默认配置是不会删除历史日志的：
<br /> status = error<br /> name = LogstashPropertiesConfig<br /> <br /> appender.console.type = Console<br /> appender.console.name = plain_console<br /> appender.console.layout.type = PatternLayout<br /> appender.console.layout.pattern = [%d{ISO8601}][%-5p][%-25c] %m%n<br /> <br /> appender.json_console.type = Console<br /> appender.json_console.name = json_console<br /> appender.json_console.layout.type = JSONLayout<br /> appender.json_console.layout.compact = true<br /> appender.json_console.layout.eventEol = true<br /> <br /> appender.rolling.type = RollingFile<br /> appender.rolling.name = plain_rolling<br /> appender.rolling.fileName = ${sys:ls.logs}/logstash-${sys:ls.log.format}.log<br /> appender.rolling.filePattern = ${sys:ls.logs}/logstash-${sys:ls.log.format}-%d{yyyy-MM-dd}.log<br /> appender.rolling.policies.type = Policies<br /> appender.rolling.policies.time.type = TimeBasedTriggeringPolicy<br /> appender.rolling.policies.time.interval = 1<br /> appender.rolling.policies.time.modulate = true<br /> appender.rolling.layout.type = PatternLayout<br /> appender.rolling.layout.pattern = [%d{ISO8601}][%-5p][%-25c] %-.10000m%n<br />
需手动加上：
<br /> appender.rolling.strategy.type = DefaultRolloverStrategy<br /> appender.rolling.strategy.action.type = Delete<br /> appender.rolling.strategy.action.basepath = ${sys:ls.logs}<br /> appender.rolling.strategy.action.condition.type = IfFileName<br /> appender.rolling.strategy.action.condition.glob = ${sys:ls.logs}/logstash-${sys:ls.log.format}<br /> appender.rolling.strategy.action.condition.nested_condition.type = IfLastModified<br /> appender.rolling.strategy.action.condition.nested_condition.age = 7D<br />

kibana

在kibana的配置文件中，只有以下几个选项：
<br /> logging.dest:<br /> Default: stdout Enables you specify a file where Kibana stores log output.<br /> logging.quiet:<br /> Default: false Set the value of this setting to true to suppress all logging output other than error messages.<br /> logging.silent:<br /> Default: false Set the value of this setting to true to suppress all logging output.<br /> logging.verbose:<br /> Default: false Set the value of this setting to true to log all events, including system usage information and all requests. Supported on Elastic Cloud Enterprise.<br /> logging.timezone<br /> Default: UTC Set to the canonical timezone id (e.g. US/Pacific) to log events using that timezone. A list of timezones can be referenced at <a href="https://en.wikipedia.org/wiki/List_of_tz_database_time_zones." rel="nofollow" target="_blank">https://en.wikipedia.org/wiki/ ... ones.</a><br />
我们可以指定输出的日志文件与日志内容，但是却不可以配置日志的rotate。这时，我们需要使用logrotate，这个linux默认安装的工具。
首先，我们要在配置文件里面指定生成pid文件：
<br /> pid.file: "pid.log"<br />
然后，修改/etc/logrotate.conf:
<br /> /var/log/kibana {<br />     missingok<br />     notifempty<br />     sharedscripts<br />     daily<br />     rotate 7<br />     copytruncate<br />     /bin/kill -HUP $(cat /usr/share/kibana/pid.log 2>/dev/null) 2>/dev/null<br />     endscript<br /> }<br />

kafka

如果不想写脚本清理过多的文件的话，需要修改config/log4j.properties文件。使用RollingFileAppender代替DailyRollingFileAppender，同时设置MaxFileSize和MaxBackupIndex。即修改为：
```

Licensed to the Apache Software Foundation (ASF) under one or more

contributor license agreements.  See the NOTICE file distributed with

this work for additional information regarding copyright ownership.

The ASF licenses this file to You under the Apache License, Version 2.0

(the "License"); you may not use this file except in compliance with

the License.  You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software

distributed under the License is distributed on an "AS IS" BASIS,

WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

See the License for the specific language governing permissions and

limitations under the License.

log4j.rootLogger=INFO, stdout

log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=[%d] %p %m (%c)%n

log4j.appender.kafkaAppender=org.apache.log4j.RollingFileAppender
log4j.appender.kafkaAppender.File=${kafka.logs.dir}/server.log
log4j.appender.kafkaAppender.MaxFileSize=10MB
log4j.appender.kafkaAppender.MaxBackupIndex=10
log4j.appender.kafkaAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.kafkaAppender.layout.ConversionPattern=[%d] %p %m (%c)%n

log4j.appender.stateChangeAppender=org.apache.log4j.RollingFileAppender
log4j.appender.stateChangeAppender.File=${kafka.logs.dir}/state-change.log
log4j.appender.stateChangeAppender.MaxFileSize=10M
log4j.appender.stateChangeAppender.MaxBackupIndex=10
log4j.appender.stateChangeAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.stateChangeAppender.layout.ConversionPattern=[%d] %p %m (%c)%n

log4j.appender.requestAppender=org.apache.log4j.RollingFileAppender
log4j.appender.requestAppender.File=${kafka.logs.dir}/kafka-request.log
log4j.appender.requestAppender.MaxFileSize=10MB
log4j.appender.requestAppender.MaxBackupIndex=10
log4j.appender.requestAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.requestAppender.layout.ConversionPattern=[%d] %p %m (%c)%n

log4j.appender.cleanerAppender=org.apache.log4j.RollingFileAppender
log4j.appender.cleanerAppender.File=${kafka.logs.dir}/log-cleaner.log
log4j.appender.cleanerAppender.MaxFileSize=10MB
log4j.appender.cleanerAppender.MaxBackupIndex=10
log4j.appender.cleanerAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.cleanerAppender.layout.ConversionPattern=[%d] %p %m (%c)%n

log4j.appender.controllerAppender=org.apache.log4j.RollingFileAppender
log4j.appender.controllerAppender.File=${kafka.logs.dir}/controller.log
log4j.appender.controllerAppender.MaxFileSize=10MB
log4j.appender.controllerAppender.MaxBackupIndex=10
log4j.appender.controllerAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.controllerAppender.layout.ConversionPattern=[%d] %p %m (%c)%n

log4j.appender.authorizerAppender=org.apache.log4j.RollingFileAppender
log4j.appender.authorizerAppender.File=${kafka.logs.dir}/kafka-authorizer.log
log4j.appender.authorizerAppender.MaxFileSize=10MB
log4j.appender.authorizerAppender.MaxBackupIndex=10
log4j.appender.authorizerAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.authorizerAppender.layout.ConversionPattern=[%d] %p %m (%c)%n

Turn on all our debugging info

log4j.logger.kafka.producer.async.DefaultEventHandler=DEBUG, kafkaAppender

log4j.logger.kafka.client.ClientUtils=DEBUG, kafkaAppender

log4j.logger.kafka.perf=DEBUG, kafkaAppender

log4j.logger.kafka.perf.ProducerPerformance$ProducerThread=DEBUG, kafkaAppender

log4j.logger.org.I0Itec.zkclient.ZkClient=DEBUG

log4j.logger.kafka=INFO, kafkaAppender

log4j.logger.kafka.network.RequestChannel$=WARN, requestAppender
log4j.additivity.kafka.network.RequestChannel$=false

log4j.logger.kafka.network.Processor=TRACE, requestAppender

log4j.logger.kafka.server.KafkaApis=TRACE, requestAppender

log4j.additivity.kafka.server.KafkaApis=false

log4j.logger.kafka.request.logger=WARN, requestAppender
log4j.additivity.kafka.request.logger=false

log4j.logger.kafka.controller=TRACE, controllerAppender
log4j.additivity.kafka.controller=false

log4j.logger.kafka.log.LogCleaner=INFO, cleanerAppender
log4j.additivity.kafka.log.LogCleaner=false

log4j.logger.state.change.logger=TRACE, stateChangeAppender
log4j.additivity.state.change.logger=false

Change this to debug to get the actual audit log for authorizer.

log4j.logger.kafka.authorizer.logger=WARN, authorizerAppender
log4j.additivity.kafka.authorizer.logger=false
```

使用过Kibana的同学应该都知道，当我们在kibana的配置文件中打开或者关闭功能，或者安装、卸载额外的插件后，重启kibana会触发一个优化的过程（optimize），如下图：

这个过程或长或短，视你电脑的性能而定。这里简单介绍一下该过程所要完成的事情。

Kibana是一个单页Web应用

首先，Kibana是一个单页的web应用。何为单页web应用？即所有的页面的读取都是在浏览器上完成，而与后台服务器无关。与后台服务器的通信只关乎数据，而非页面。所以，应用上所有的UI都被打包在一起，一次性的发送到了浏览器端，而不是通过URL到后台进行获取。所以，我们看到kibana的首页是下面这样的：
<a href="http://localhost:5601/app/kibana#/" rel="nofollow" target="_blank">http://localhost:5601/app/kibana#/</a>
注意这里的#后，代表#后面的内容会被浏览器提取，不往服务器端进行url的情况，而是在浏览器上进行内部重新渲染。因为所有的页面都是存储在浏览器的，所有在初次访问的时候，会加载大量的代码到浏览器端，这些代码都是被压缩过的bundle文件：

而optimize的过程，就是把这些原本可读性的源代码压缩为bundle.js的过程。因此，每当你对Kibana进行裁剪之后重启，因为前端的部分是完全由浏览器负责的，所有bundle文件需要重新生成后再发给浏览器，所以会触发optimize的过程。

Kibana在6.2.0版本之后，常规版本已经默认自带了xpack（当然，你还是可以直接下载不带xpack的开源社区版），导致Kibana的size已经到了200M左右，而且越往后的版本，功能越多，代码量越大，每次optimize的过程都会耗费更多的时间。一般来说，我们会将Kibana部署在单独的机器上，因为这仅仅是一个web后端，通常我们不会分配比较优质的资源，（2C4G都算浪费的了），这种情况下面，每次我们裁剪后重启Kibana都会耗费半个小时～1个小时的时间，更有甚者直接hang住，查看系统日志才知道OOM了。

Nodejs的内存机制

Kibana是用Nodejs编写的程序，在一般的后端语言中，基本的内存使用上基本没有什么限制，但是在nodeJs中却只能使用部分内存。在64位系统下位约为1.4G，在32位系统下约为0.7G，造成这个问题的主要原因是因为nodeJs基于V8构建，V8使用自己的方式来管理和分配内存，这一套管理方式在浏览器端使用绰绰有余，但是在nodeJs中这却限制了开发者，在应用中如果碰到了这个限制，就会造成进程退出。

Nodejs内存机制对Kibana优化的影响

因为Kibana的代码体量越来越大，将所有的代码加载到内存之后，再解析语法树，进行bundle的转换所耗费的内存已经接近1.4G的限制了，当你安装更多插件，比如sentinl的时候，系统往往已经无法为继，导致Kibana无法启动

解决方案

这种情况下，我们需要在Kibana启动的时候，指定NodeJs使用更多的内存。这个可以通过设置Node的环境变量办到。
<br /> NODE_OPTIONS="--max-old-space-size=4096"<br />
当然，我的建议是直接指定在kibana的启动脚本当中，修改/usr/share/kibana/bin/kibana文件为：
```shell

!/bin/sh

SCRIPT=$0

SCRIPT may be an arbitrarily deep series of symlinks. Loop until we have the concrete path.

while [ -h "$SCRIPT" ] ; do
ls=$(ls -ld "$SCRIPT")

Drop everything prior to ->

link=$(expr "$ls" : '.-> (.)$')
if expr "$link" : '/.*' > /dev/null; then
SCRIPT="$link"
else
SCRIPT=$(dirname "$SCRIPT")/"$link"
fi
done

DIR="$(dirname "${SCRIPT}")/.."
NODE="${DIR}/node/bin/node"
test -x "$NODE" || NODE=$(which node)
if [ ! -x "$NODE" ]; then
echo "unable to find usable node.js executable."
exit 1
fi

NODE_ENV=production exec "${NODE}" $NODE_OPTIONS --max_old_space_size=3072 --no-warnings "${DIR}/src/cli" ${@}

``<br /> 改动在最后一句：NODE_ENV=production exec "${NODE}" $NODE_OPTIONS --max_old_space_size=3072 --no-warnings "${DIR}/src/cli" ${@}`

这样，我们可以保证Kibana能顺利的完成optimize的过程