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);

	}   

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/downloads/elasticsearch-plugins/repository-hdfs/repository-hdfs-5.0.2.zip 保存在/usr/local下

• 安装插件

  cd /usr/local/es/elasticsearch-5.0.2/bin
  ./elasticsearch-plugin install file:///usr/local/repository-hdfs-5.0.2.zip

• 安装成功后需要重启Elasticsearch

备份与恢复

• 构建一个仓库

  PUT http://192.168.10.74:9200/_snapshot/backup
  {  
  "type": "hdfs",  
    "settings": {  
            "uri": "hdfs://192.168.10.170:9000",  
            "path": "/es",  
            "conf_location": "/usr/local/hadoop/etc/hadoop/hdfs-site.xml"  
    }
  }

• 备份快照

  PUT http://192.168.10.74:9200/_snapshot/backup/snapshot_users?wait_for_completion=true
  {
  "indices": "users",  //备份users的index，注意不设置这个属性，默认是备份所有index
  "ignore_unavailable": true,
  "include_global_state": false
  }

• 恢复快照

  POST http://192.168.10.74:9200/_snapshot/backup/snapshot_users/_restore
  {
  "indices": "users",    //指定索引恢复，不指定就是所有
  "ignore_unavailable": true,     //忽略恢复时异常索引
  "include_global_state": false    //是否存储全局转态信息,fasle代表有一个或几个失败，不会导致整个任务失败
  }

整合Spark与Elasticsearch

整体思路

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

引入依赖

<dependency>
    <groupId>org.elasticsearch</groupId>
    <artifactId>elasticsearch-hadoop</artifactId>
    <version>5.0.2</version>
</dependency>

具体流程

• 数据在HDFS上，数据存储在HDFS的每个DataNode的block上

• 数据加载到Spark SQL

  • 数据从HDFS加载到Spark SQL中，以RDD形式存储

  JavaRDD<String> textFile = spark.read().textFile("hdfs://192.168.10.170:9000/csv/user.csv")

  • 添加数据结构信息转换为新的RDD

  JavaRDD<UserItem> dataSplits = textFile.map(line -> {
      String records = line.toString().trim();
      String record = records.substring(0,records.length() - 1).trim();
      String[] parts = record.split("\\|");
      UserItem u = new UserItem();
      u.setName(parts[0]);
      u.setAge(parts[1]);
      u.setHeight(parts[2]);
      return u;
  });

  • 根据新的RDD创建DataFrame

  DataSet<Row> ds = spark.createDataFrame(dataSplits, UserItem.class);

• 由Dataset创建索引，并写入ES

  JavaEsSparkSQL.saveToEs(ds, "es_spark/users");

• 数据在ES中建立索引

• Spark SQL通过索引对ES中的数据进行查询

  SparkSession spark = SparkSession.builder().appName("es-spark").master("local").config("es.index.auto.create", true).getOrCreate();
  Map<String, String> options = new HashMap<>();
  options.put("pushdown", "true");
  options.put("es.nodes","192.168.10.74:9200");
  
  Dataset<Row> df = spark.read().options(options).format("org.elasticsearch.spark.sql").load("es_spark/users");
  df.createOrReplaceTempView("users");
  
  Dataset<Row> userSet = spark.sql("SELECT name FORM users WHERE age >=10 AND age <= 20");
  userSet.show();

结束

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

{

  "仓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，这里不展开介绍。

3. 日志场景调优

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

3.1 基础场景

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

• 建议打开最优压缩，一般可降低40%存储。
• 设置原始日志字段（message）为text，去除keyword类型子字段，提供全文搜索能力，降低存储。
• 关闭_all索引，前面已通过message提供全文搜索能力。
• 对于其他字符串字段，统一设置为keyword类型，避免默认情况下字符串字段同时存储text、keyword两种类型的数据。
• 使用开源组件（如Beats）上报数据时会包含较多辅助信息，用户可通过修改组件配置文件进行裁剪。

这样去除message的keyword子字段、_all等冗余信息后，再加上最优压缩，可以保证数据相对精简。下面给出这类场景的常用模板，供用户参考：

{
    "order": 5,
    "template": "my_log_*",
    "settings": {
        "translog.durability": "async",
        "translog.sync_interval": "5s",
        "index.refresh_interval": "30s",
        "index.codec": "best_compression"    # 最优压缩
    },
    "mappings": {
        "_default_": {
            "_all": {                        # 关闭_all索引
                "enabled": false
            },
            "dynamic_templates": [
                {
                    "log": {                 # 原始日志字段，分词建立索引
                        "match": "message",
                        "mapping": {
                            "type": "text"
                        }
                    }
                },
                {
                    "strings": {             # 其他字符串字段，统一设置为keyword类型
                        "match_mapping_type": "string",
                        "mapping": {
                            "type": "keyword"
                        }
                    }
                }
            ]
        }
    }
}

3.2 精准搜索场景

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

• 清洗过程中，可仅解析出需要精准搜索的部分作为独立字段，用于精准搜索。
• 对于精准搜索字段，如果无排序/聚合需求，可以关闭doc_values；对于字符串，一般使用keyword，可按需考虑使用text。

下面给出这类场景的常用模板，供用户参考：

{
    "order": 5,
    "template": "my_log_*",
    "settings": {
        "translog.durability": "async",
        "translog.sync_interval": "5s",
        "index.refresh_interval": "30s",
        "index.codec": "best_compression"    # 最优压缩
    },
    "mappings": {
        "_default_": {
            "_all": {                        # 关闭_all索引
                "enabled": false
            },
            "dynamic_templates": [
                {
                    "log": {                 # 原始日志字段，分词建立索引
                        "match": "message",
                        "mapping": {
                            "type": "text"
                        }
                    }
                },
                {
                    "precise_fieldx": {       # 精准搜索字段
                        "match": "fieldx",
                        "mapping": {
                            "type": "keyword",
                            "doc_values": false
                        }
                    }
                },
                {
                    "strings": {             # 其他字符串字段，统一设置为keyword类型
                        "match_mapping_type": "string",
                        "mapping": {
                            "type": "keyword"
                        }
                    }
                }
            ]
        }
    }
}

3.3 统计分析场景

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

• 清洗过程中，解析出所有需要的数据作为独立字段；原始日志非必要时，建议去除。
• 如果有强需求保留原始日志，可以设置该字段enabled属性为false，只存储不索引。
• 多数字段保持默认即可，会自动建立索引、打开doc_values，可用于查询、排序、聚合。
• 对部分无排序/聚合需求、开销高的字段，可以关闭doc_values。

下面给出这类场景的常用模板，供用户参考：

{
    "order": 5,
    "template": "my_log_*",
    "settings": {
        "translog.durability": "async",
        "translog.sync_interval": "5s",
        "index.refresh_interval": "30s",
        "index.codec": "best_compression"    # 最优压缩
    },
    "mappings": {
        "_default_": {
            "_all": {                        # 关闭_all索引
                "enabled": false
            },
            "dynamic_templates": [
                {
                    "log": {                 # 原始日志字段，关闭索引
                        "match": "message",
                        "mapping": {
                            "enabled": false
                        }
                    }
                },
                {
                    "index_only_fieldx": {   # 仅索引的字段，无排序/聚合需求
                        "match": "fieldx",
                        "mapping": {
                            "type": "keyword",
                            "doc_values": false
                        }
                    }
                },
                {
                    "strings": {             # 其他字符串字段，统一设置为keyword类型
                        "match_mapping_type": "string",
                        "mapping": {
                            "type": "keyword"
                        }
                    }
                }
            ]
        }
    }
}

ES 5.1及之后的版本，支持关键字查询时自动选择目标字段，用户没有必要再使用原始日志字段提供不指定字段进行查询的能力。

4. 小结

日志的使用方式比较灵活，本文结合常见的客户使用方式，从整体上对性能、成本进行优化。用户也可结合自身业务场景，参考文章Elasticsearch调优实践进行更细致的优化。

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

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

安装和配置metricbeat

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

# Module: kafka
# Docs: https://www.elastic.co/guide/en/beats/metricbeat/6.4/metricbeat-module-kafka.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：

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.DailyRollingFileAppender
log4j.appender.kafkaAppender.DatePattern='.'yyyy-MM-dd-HH
log4j.appender.kafkaAppender.File=${kafka.logs.dir}/server.log
log4j.appender.kafkaAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.kafkaAppender.layout.ConversionPattern=[%d] %p %m (%c)%n

log4j.appender.stateChangeAppender=org.apache.log4j.DailyRollingFileAppender
log4j.appender.stateChangeAppender.DatePattern='.'yyyy-MM-dd-HH
log4j.appender.stateChangeAppender.File=${kafka.logs.dir}/state-change.log
log4j.appender.stateChangeAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.stateChangeAppender.layout.ConversionPattern=[%d] %p %m (%c)%n

log4j.appender.requestAppender=org.apache.log4j.DailyRollingFileAppender
log4j.appender.requestAppender.DatePattern='.'yyyy-MM-dd-HH
log4j.appender.requestAppender.File=${kafka.logs.dir}/kafka-request.log
log4j.appender.requestAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.requestAppender.layout.ConversionPattern=[%d] %p %m (%c)%n

log4j.appender.cleanerAppender=org.apache.log4j.DailyRollingFileAppender
log4j.appender.cleanerAppender.DatePattern='.'yyyy-MM-dd-HH
log4j.appender.cleanerAppender.File=${kafka.logs.dir}/log-cleaner.log
log4j.appender.cleanerAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.cleanerAppender.layout.ConversionPattern=[%d] %p %m (%c)%n

log4j.appender.controllerAppender=org.apache.log4j.DailyRollingFileAppender
log4j.appender.controllerAppender.DatePattern='.'yyyy-MM-dd-HH
log4j.appender.controllerAppender.File=${kafka.logs.dir}/controller.log
log4j.appender.controllerAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.controllerAppender.layout.ConversionPattern=[%d] %p %m (%c)%n

log4j.appender.authorizerAppender=org.apache.log4j.DailyRollingFileAppender
log4j.appender.authorizerAppender.DatePattern='.'yyyy-MM-dd-HH
log4j.appender.authorizerAppender.File=${kafka.logs.dir}/kafka-authorizer.log
log4j.appender.authorizerAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.authorizerAppender.layout.ConversionPattern=[%d] %p %m (%c)%n

解决方案

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

elasticsearch 

通过修改log4j2.properties文件来解决。该文件在/etc/elasticsesarch目录下（或者config目录）。 默认配置是：

appender.rolling.type = RollingFile 
appender.rolling.name = rolling
appender.rolling.fileName = ${sys:es.logs.base_path}${sys:file.separator}${sys:es.logs.cluster_name}.log 
appender.rolling.layout.type = PatternLayout
appender.rolling.layout.pattern = [%d{ISO8601}][%-5p][%-25c{1.}] [%node_name]%marker %.-10000m%n
appender.rolling.filePattern = ${sys:es.logs.base_path}${sys:file.separator}${sys:es.logs.cluster_name}-%d{yyyy-MM-dd}-%i.log.gz 
appender.rolling.policies.type = Policies
appender.rolling.policies.time.type = TimeBasedTriggeringPolicy 
appender.rolling.policies.time.interval = 1 
appender.rolling.policies.time.modulate = true 
appender.rolling.policies.size.type = SizeBasedTriggeringPolicy 
appender.rolling.policies.size.size = 256MB 
appender.rolling.strategy.type = DefaultRolloverStrategy
appender.rolling.strategy.fileIndex = nomax
appender.rolling.strategy.action.type = Delete 
appender.rolling.strategy.action.basepath = ${sys:es.logs.base_path}
appender.rolling.strategy.action.condition.type = IfFileName 
appender.rolling.strategy.action.condition.glob = ${sys:es.logs.cluster_name}-* 
appender.rolling.strategy.action.condition.nested_condition.type = IfAccumulatedFileSize 
appender.rolling.strategy.action.condition.nested_condition.exceeds = 2GB 

以上默认配置，会保存2GB的日志，只有累计的日志大小超过2GB的时候，才会删除旧的日志文件。 建议改为如下配置，仅保留最近7天的日志

appender.rolling.strategy.type = DefaultRolloverStrategy
appender.rolling.strategy.action.type = Delete
appender.rolling.strategy.action.basepath = ${sys:es.logs.base_path}
appender.rolling.strategy.action.condition.type = IfFileName
appender.rolling.strategy.action.condition.glob = ${sys:es.logs.cluster_name}-*
appender.rolling.strategy.action.condition.nested_condition.type = IfLastModified
appender.rolling.strategy.action.condition.nested_condition.age = 7D

这里必须注意，log4j2会因为末尾的空格导致无法识别配置

logstash

与elasticsearch类似，通过修改log4j2.properties文件来解决。该文件在/etc/logstash目录下（或者config目录）。 默认配置是不会删除历史日志的：

status = error
name = LogstashPropertiesConfig

appender.console.type = Console
appender.console.name = plain_console
appender.console.layout.type = PatternLayout
appender.console.layout.pattern = [%d{ISO8601}][%-5p][%-25c] %m%n

appender.json_console.type = Console
appender.json_console.name = json_console
appender.json_console.layout.type = JSONLayout
appender.json_console.layout.compact = true
appender.json_console.layout.eventEol = true

appender.rolling.type = RollingFile
appender.rolling.name = plain_rolling
appender.rolling.fileName = ${sys:ls.logs}/logstash-${sys:ls.log.format}.log
appender.rolling.filePattern = ${sys:ls.logs}/logstash-${sys:ls.log.format}-%d{yyyy-MM-dd}.log
appender.rolling.policies.type = Policies
appender.rolling.policies.time.type = TimeBasedTriggeringPolicy
appender.rolling.policies.time.interval = 1
appender.rolling.policies.time.modulate = true
appender.rolling.layout.type = PatternLayout
appender.rolling.layout.pattern = [%d{ISO8601}][%-5p][%-25c] %-.10000m%n

需手动加上：

appender.rolling.strategy.type = DefaultRolloverStrategy
appender.rolling.strategy.action.type = Delete
appender.rolling.strategy.action.basepath = ${sys:ls.logs}
appender.rolling.strategy.action.condition.type = IfFileName
appender.rolling.strategy.action.condition.glob = ${sys:ls.logs}/logstash-${sys:ls.log.format}
appender.rolling.strategy.action.condition.nested_condition.type = IfLastModified
appender.rolling.strategy.action.condition.nested_condition.age = 7D

kibana

在kibana的配置文件中，只有以下几个选项：

logging.dest:
Default: stdout Enables you specify a file where Kibana stores log output.
logging.quiet:
Default: false Set the value of this setting to true to suppress all logging output other than error messages.
logging.silent:
Default: false Set the value of this setting to true to suppress all logging output.
logging.verbose:
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.
logging.timezone
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 https://en.wikipedia.org/wiki/List_of_tz_database_time_zones.

我们可以指定输出的日志文件与日志内容，但是却不可以配置日志的rotate。这时，我们需要使用logrotate，这个linux默认安装的工具。 首先，我们要在配置文件里面指定生成pid文件：

pid.file: "pid.log"

然后，修改/etc/logrotate.conf:

/var/log/kibana {
    missingok
    notifempty
    sharedscripts
    daily
    rotate 7
    copytruncate
    /bin/kill -HUP $(cat /usr/share/kibana/pid.log 2>/dev/null) 2>/dev/null
    endscript
}

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的首页是下面这样的： http://localhost:5601/app/kibana#/ 注意这里的#后，代表#后面的内容会被浏览器提取，不往服务器端进行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的环境变量办到。

NODE_OPTIONS="--max-old-space-size=4096"

当然，我的建议是直接指定在kibana的启动脚本当中，修改/usr/share/kibana/bin/kibana文件为：

#!/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" ${@}

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

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