ELK Tips 主要介绍一些 ELK 使用过程中的小技巧，内容主要来源为 Elastic 中文社区。

一、Logstash

1、Filebeat ：Non-zero metrics in the last 30s

• 问题表现：Filebeat 无法向 Elasticsearch 发送日志数据；
  
• 错误信息：INFO [monitoring] 1og/log.go:124 Non-zero metrics in the last 30s；
  
• 社区反馈：在 input 和 output 下面添加属性 enabled：true。
  
  ```
  filebeat.inputs:
  
• type: log
  enabled: true
  paths:
  
• /var/log/*.log
  
  output.elasticsearch:
  hosts: ["<a href="https://localhost:9200"" rel="nofollow" target="_blank">https://localhost:9200"]
  username: "filebeat_internal"
  password: "YOUR_PASSWORD"
  enabled: true
  ```
  
  input 和 output 下 enabled 属性默认值为 true，因此怀疑另有其因。
  
  

  2、Logstash 按月生成索引
  

  <br /> output {<br /> if [type] == "typeA"{<br /> elasticsearch {<br /> hosts => "127.0.0.1:9200"<br /> index => "log_%{+YYYY_MM}"<br /> }<br /> }<br /> }<br />
  
  按照日的原理类似：%{+YYYY.MM.dd}
  
  

  3、Filebeat 通过配置删除特定字段
  

  Filebeat 实现了类似 Logstash 中 filter 的功能，叫做处理器（processors），processors 种类不多，尽可能在保持 Filebeat 轻量化的基础上提供更多常用的功能。
  
  下面列几种常用的 processors：
  

• add_cloud_metadata：添加云服务器的 meta 信息；
  
• add_locale：添加本地时区；
  
• decode_json_fields：解析并处理包含 Json 字符串的字段；
  
• drop_event：丢弃符合条件的消息事件；
  
• drop_fields：删除符合条件的字段；
  
• include_fields：选择符合条件的字段；
  
• rename：字段重命名；
  
• add_kubernetes_metadata：添加 k8s 的 meta 信息；
  
• add_docker_metadata：添加容器的 meta 信息；
  
• add_host_metadata：添加操作系统的 meta 信息；
  
• dissect：类似与 gork 的正则匹配字段的功能；
  
• dns：配置 filebeat 独立的 dns 解析方式；
  
• add_process_metadata：添加进程的元信息。
  
  processors 的使用方式：
  ```
  
• type:
  processors:
  
• :
  when:
  
  
  ...
  ```
  
  ### 4、LogStash 采集 FTP 日志文件
  ```
  exec {
  codec => plain { }
  command => "curl ftp://server/logs.log"
  interval => 3000}
  }
  ```
  
  ### 5、Logstash docker-compose 启动失败（Permission denied）
  在 docker-compose 中使用 user 选项设置使用 root 用户启动 docker，能解决权限问题。
  ```
  $ cat docker-compose.yml
  
  version: '2'
  services:
  logstash:
  image: docker.elastic.co/logstash/logstash:6.4.2
  user: root
  command: id
  ```
  
  ### 6、Metricize filter plugin
  将一条消息拆分为多条消息。
  
  ```
  # 原始信息
  {
  type => "type A"
  metric1 => "value1"
  metric2 => "value2"
  }
  
  # 配置信息
  filter {
  metricize {
  metrics => [ "metric1", "metric2" ]
  }
  }
  
  # 最终输出
  { {
  type => "type A" type => "type A"
  metric => "metric1" metric => "metric2"
  value => "value1" value => "value2"
  } }
  ```
  
  ## 二、Elasticsearch
  ### 1、ES 倒排索引内部结构
  Lucene 的倒排索引都是按照字段（field）来存储对应的文档信息的，如果 docName 和 docContent 中有“苹果”这个 term，就会有这两个索引链，如下所示：
  ```
  docName：
  "苹果" -> "doc1, doc2, doc3..."
  
  docContent：
  "苹果" -> "doc2, doc4, doc6..."
  ```
  
  ### 2、Jest 和 RestHighLevelClient 哪个好用点
  RestHighLevelClient 是官方组件，会一直得到官方的支持，且会与 ES 保持同步更新，推荐使用官方的高阶 API。
  
  Jest 由于是社区维护，所以更新会有一定延迟，目前最新版对接 ES6.3.1，近一个月只有四个 issue，说明整体活跃度较低，因此不推荐使用。
  
  此外推荐一份 TransportClient 的中文使用手册，翻译的很不错：[https://github.com/jackiehff/e ... pi-cn](https://github.com/jackiehff/e ... api-cn)。
  
  ### 3、ES 单分片使用 From/Size 分页遇到重复数据
  常规情况下 ES 单分片使用 From/Size 是不会遇到数据重复的，数据重复的可能原因有：
  
• 没有添加排序；
  
• 添加了按得分排序，但是查询语句全部为 filter 过滤条件（此时得分都一致）；
  
• 添加了排序，但是有索引中文档的新增、修改、删除等操作。
  
  对于多分片，推荐添加 preference 参数来实现分页结果的一致性。
  
  

  4、The number of object passed must be even but was [1]
  

  ES 在调用 setSource 的时候传入 Json 对象后会报错：The number of object passed must be even but was [1]，此时可以推荐将 Json 对象转为 Map 集合，或者把 Json 对象转为 json 字符串，不过传入字符串的时候需要设置类型。
  <br /> IndexRequest indexRequest = new IndexRequest("index", "type", "id");<br /> JSONObject doc = new JSONObject();<br /> //indexRequest.source(jsonObject); 错误的使用方法<br /> //转为 Map 对象<br /> indexRequest.source(JSONObject.parseObject((String) doc.get("json"), Map.class));<br /> //转为 Json 字符串（声明字符串类型）<br /> indexRequest.source(JSON.toJSONString(doc), XContentType.JSON);<br />
  
  

  5、跨集群搜索
  

  ES 6.X 原生支持跨集群搜索，具体配置请参考：https://www.elastic.co/guide/e ... .html
  <br /> PUT _cluster/settings<br /> {<br /> "persistent": {<br /> "cluster": {<br /> "remote": {<br /> "cluster_one": {<br /> "seeds": [<br /> "127.0.0.1:9300"<br /> ]<br /> },<br /> "cluster_two": {<br /> "seeds": [<br /> "127.0.0.1:9301"<br /> ]<br /> },<br /> "cluster_three": {<br /> "seeds": [<br /> "127.0.0.1:9302"<br /> ]<br /> }<br /> }<br /> }<br /> }<br /> }<br />
  
  ES 6.5 推出了新功能，跨集群同步（Cross-cluster replication），感兴趣的可以自行了解。
  
  

  6、ES 排序时设置空值排序位置
  

  <br /> GET /_search<br /> {<br /> "sort" : [<br /> { "price" : {"missing" : "_last"} }<br /> ],<br /> "query" : {<br /> "term" : { "product" : "chocolate" }<br /> }<br /> }<br />
  
  

  7、ES 冷归档数据如何处理
  

  使用相对低配的大磁盘机器配置为 ES 的 Warm Nodes，可以通过 index.routing.allocation.require.box_type 来设置索引是冷数据或者热数据。如果索引极少使用，可以 close 索引，然后在需要搜索的时候 open 即可。
  
  

  8、ES 相似文章检测
  

  对于大文本的去重，可以参考 SimHash 算法，通过 SimHash 可以提取到文档指纹（64位），两篇文章通过 SimHash 计算海明距离即可判断是否重复。海明距离计算，可以通过插件实现：https://github.com/joway/elast ... lugin
  
  

  9、Terms 聚合查询优化
  

• 如果只需要聚合后前 N 条记录，推荐在 Terms 聚合时添加上 "collect_mode": "breadth_first"；
  
• 此外可以通过设置 "min_doc_count": 10来限制最小匹配文档数；
  
• 如果对返回的 Term 有所要求，可以通过设置 include 和 exclude 来过滤 Term；
  
• 如果想获取全部 Term 聚合结果，但是聚合结果又很多，可以考虑将聚合分成多个批次分别取回（Filtering Values with partitions）。
  
  

  10、Tomcat 字符集造成的 ES 查询无结果
  

  两个系统连接同一个 ES 服务，配置和代码完全一致，同一个搜索条件，一个能够搜索出来东西，一个什么都搜索不出来，排查结果是因为其中一个系统的 tomcat 配置有问题，导致请求的时候乱码了，所以搜不到数据。
  
  

  11、ES 索引设置默认分词器
  

  默认情况下，如果字段不指定分词器，ES 或使用 standard 分词器进行分词；可以通过下面的设置更改默认的分词器。
  
  2.X 支持设置默认的索引分词器（default_index）和默认的查询分词器（default_search），6.X 已经不再支持。
  
  <br /> PUT /index<br /> {<br /> "settings": {<br /> "analysis": {<br /> "analyzer": {<br /> "default": {<br /> "type": "ik_max_word",<br /> "tokenizer": "ik_max_word"<br /> }<br /> }<br /> }<br /> }<br /> }<br />
  
  

  12、ES 中的魔法参数
  

• 索引名：_index
  
• 类型名：_type
  
• 文档Id：_id
  
• 得分：_score
  
• 索引排序：_doc
  
  如果你对排序没有特别的需求，推荐使用 _doc 进行排序，例如执行 Scroll 操作时。
  
  

  13、ES 延迟执行数据上卷（Rollup ）
  

  Rollup job 有个 delay 参数控制 job 执行的延迟时间，默认情况下不延迟执行，这样如果某个 interval 的数据已经聚合好了，该 interval 迟到的数据是不会处理的。
  
  好在 rollup api 可以支持同时搜索裸索引和 rollup 过的索引，所以如果数据经常有延迟的话，可以考虑设置一个合适的 delay，比如 1h、6h 甚至 24h，这样 rollup 的索引产生会有延迟，但是能确保迟到的数据被处理。
  
  从应用场景上看，rollup 一般是为了对历史数据做聚合存放，减少存储空间，所以延迟几个小时，甚至几天都是合理的。搜索的时候，同时搜索最近的裸索引和历史的 rollup 索引，就能将两者的数据组合起来，在给出正确的聚合结果的情况下，又兼顾了性能。
  
  Rollup 是实验性功能，不过非常有用，特别是使用 ES 做数据仓库的场景。
  
  

  14、ES6.x 获取所有的聚合结果
  

  ES2.x 版本中，在聚合查询时，通过设置 setSize(0) 就可以获取所有的聚合结果，在ES6.x 中直接设置 setSize(Integer.MAX_VALUE) 等效于 2.x 中设置为 0。
  
  

  15、ES Jar 包冲突问题
  

  经常会遇到 ES 与业务集成时出现 Jar 包冲突问题，推荐的解决方法是使用 maven-shade-plugin 插件，该插件通过将冲突的 Jar 包更换一个命名空间的方式来解决 Jar 包的冲突问题，具体使用可以参考文章：[https://www.jianshu.com/p/d9fb7afa634d](https://www.jianshu.com/p/d9fb7afa634d)。
  ```
  

  
  
  org.apache.maven.plugins
  maven-shade-plugin
  2.4.1
  
  false
  
  
  
  package
  
  shade
  
  
  
  
  com.google.guava
  net.luculent.elasticsearch.guava
  
  
  com.fasterxml.jackson
  net.luculent.elasticsearch.jackson
  
  
  org.joda
  net.luculent.elasticsearch.joda
  
  
  com.google.common
  net.luculent.elasticsearch.common
  
  
  com.google.thirdparty
  net.luculent.elasticsearch.thirdparty
  
  
  
  
  
  
  
  
  
  
  
  ```
  ### 16、ES 如何选择 Shard 存储文档?
  ES 采用 djb2 哈希算法对要索引文档的指定（或者默认随机生成的）`_id` 进行哈希，得到哈希结果后对索引 shard 数目 n 取模，公式如下：`hash(_id) % n`；根据取模结果决定存储到哪一个 shard 。
  
  ## 三、Kibana
  ### 1、在 Kiabana 的 Discovery 界面显示自定义字段
  Kibana 的 Discovery 界面默认只显示 time 和 _source 两个字段，这个界面的左半部分，在 Popular 下面展示了很多，你只需要在你需要展示的字段后面点击 add 即可将自定义的字段添加到 discovery 界面。
  
  
  
  ### 2、filebeat 的 monitor 指标的说明
  
• Total：'All events newly created in the publishing pipeline'
  
• Emitted： 'Events processed by the output (including retries)'
  
• Acknowledged：'Events acknowledged by the output (includes events dropped by the output)'
  
• Queued：'Events added to the event pipeline queue'
  
  

  四、社区文章精选
  

• [Elastic认证考试心得](https://note.youdao.com/)
  
• [一文快速上手Logstash](https://elasticsearch.cn/article/6141)
  
• [当Elasticsearch遇见Kafka--Kafka Connect](https://elasticsearch.cn/article/6140)
  
• [elasticsearch冷热数据读写分离](https://elasticsearch.cn/article/6127)
  
• [elasticsearch优秀实践](https://elasticsearch.cn/article/6126)
  
• [ELK 使用小技巧（第 1 期）](https://elasticsearch.cn/article/1006)
  
  ---
  Any Code，Code Any！
  
  扫码关注『AnyCode』，编程路上，一起前行。
  
  

本文同步发布在腾讯云+社区Elasticsearch专栏中：https://cloud.tencent.com/developer/column/4008
在“[当Elasticsearch遇见Kafka--Logstash kafka input插件](https://cloud.tencent.com/developer/article/1362320)”一文中，我对Logstash的Kafka input插件进行了简单的介绍，并通过实际操作的方式，为大家呈现了使用该方式实现Kafka与Elastisearch整合的基本过程。可以看出使用Logstash input插件的方式，具有配置简单，数据处理方便等优点。然而使用Logstash Kafka插件并不是Kafka与Elsticsearch整合的唯一方案，另一种比较常见的方案是使用Kafka的开源组件Kafka Connect。


1 Kafka Connect简介

Kafka Connect是Kafka的开源组件Confluent提供的功能，用于实现Kafka与外部系统的连接。Kafka Connect同时支持分布式模式和单机模式，另外提供了一套完整的REST接口，用于查看和管理Kafka Connectors，还具有offset自动管理，可扩展等优点。

Kafka connect分为企业版和开源版，企业版在开源版的基础之上提供了监控，负载均衡，副本等功能，实际生产环境中建议使用企业版。(本测试使用开源版)

Kafka connect workers有两种工作模式，单机模式和分布式模式。在开发和适合使用单机模式的场景下，可以使用standalone模式, 在实际生产环境下由于单个worker的数据压力会比较大，distributed模式对负载均和和扩展性方面会有很大帮助。（本测试使用standalone模式）

关于Kafka Connect的详细情况可以参考[[Kafka Connect](https://docs.confluent.io/curr ... x.html)]

2 使用Kafka Connect连接Kafka和Elasticsearch

2.1 测试环境准备

本文与使用Logstash Kafka input插件环境一样[[传送门](https://cloud.tencent.com/developer/article/1362320)]，组件列表如下

| 服务 | ip | port |
|:----|:----|:----|
| Elasticsearch service | 192.168.0.8 | 9200 |
| Ckafka | 192.168.13.10 | 9092 |
| CVM | 192.168.0.13 | - |


kafka topic也复用原来了的kafka_es_test

2.2 Kafka Connect 安装

[[Kafka Connec下载地址](https://www.confluent.io/download/)]

本文下载的为开源版本confluent-oss-5.0.1-2.11.tar.gz，下载后解压

2.3 Worker配置

1) 配置参考

如前文所说,worker分为Standalone和Distributed两种模式，针对两种模式的配置，参考如下

[[通用配置](https://www.confluent.io/download/)]

[[Standalone Woker配置](https://www.confluent.io/download/)]

[[Distributed Worker配置](https://www.confluent.io/download/)]

此处需要注意的是Kafka Connect默认使用AvroConverter，使用该AvroConverter时需要注意必须启动Schema Registry服务

2) 实际操作

本测试使用standalone模式，因此修改/root/confluent-5.0.1/etc/schema-registry/connect-avro-standalone.properties

<br /> bootstrap.servers=192.168.13.10:9092<br />

2.4 Elasticsearch Connector配置

1) 配置参考

[[Connectors通用配置](https://docs.confluent.io/curr ... ectors)]

[[Elasticsearch Configuration Options](https://docs.confluent.io/curr ... s.html)]

2) 实际操作

修改/root/confluent-5.0.1/etc/kafka-connect-elasticsearch/quickstart-elasticsearch.properties

<br /> name=elasticsearch-sink<br /> connector.class=io.confluent.connect.elasticsearch.ElasticsearchSinkConnector<br /> tasks.max=1<br /> topics=kafka_es_test<br /> key.ignore=true<br /> connection.url=http://192.168.0.8:9200<br /> type.name=kafka-connect<br />

注意: 其中topics不仅对应Kafka的topic名称，同时也是Elasticsearch的索引名，当然也可以通过topic.index.map来设置从topic名到Elasticsearch索引名的映射

2.5 启动connector

1 注意事项

1) 由于配置文件中jar包位置均采用的相对路径，因此建议在confluent根目录下执行命令和启动程序，以避免不必要的问题

2) 如果前面没有修改converter，仍采用AvroConverter, 注意需要在启动connertor前启动Schema Registry服务

2 启动Schema Registry服务

正如前文所说，由于在配置worker时指定使用了AvroConverter，因此需要启动Schema Registry服务。而该服务需要指定一个zookeeper地址或Kafka地址，以存储schema数据。由于CKafka不支持用户通过接口形式创建topic，因此需要在本机起一个kafka以创建名为_schema的topic。

1) 启动Zookeeper

<br /> ./bin/zookeeper-server-start -daemon etc/kafka/zookeeper.properties<br />

2) 启动kafka

<br /> ./bin/kafka-server-start -daemon etc/kafka/server.properties<br />

3) 启动schema Registry

<br /> ./bin/schema-registry-start -daemon etc/schema-registry/schema-registry.properties<br />

4) 使用netstat -natpl 查看各服务端口是否正常启动

zookeeper 2181

kafka 9092

schema registry 8081

3 启动Connector

<br /> ./bin/connect-standalone -daemon etc/schema-registry/connect-avro-standalone.properties etc/kafka-connect-elasticsearch/quickstart-elasticsearch.properties<br />

ps：以上启动各服务均可在logs目录下找到对应日志

2.6 启动Kafka Producer

由于我们采用的是AvroConverter，因此不能采用Kafka工具包中的producer。Kafka Connector bin目录下提供了Avro Producer

1) 启动Producer

<br /> ./bin/kafka-avro-console-producer --broker-list 192.168.13.10:9092 --topic kafka_es_test --property value.schema='{"type":"record","name":"person","fields":[{"name":"nickname","type":"string"}]}'<br />

2) 输入如下数据

<br /> {"nickname":"michel"}<br /> {"nickname":"mushao"}<br />

2.7 Kibana验证结果

1) 查看索引

在kibana Dev Tools的Console中输入

<br /> GET _cat/indices<br />

结果

<br /> green open kafka_es_test 36QtDP6vQOG7ubOa161wGQ 5 1 1 0 7.9kb 3.9kb<br /> green open .kibana QUw45tN0SHqeHbF9-QVU6A 1 1 1 0 5.5kb 2.7kb<br />

可以看到名为kafka_es_test的索引被成功创建

2) 查看数据

<br /> {<br /> "took": 0,<br /> "timed_out": false,<br /> "_shards": {<br /> "total": 5,<br /> "successful": 5,<br /> "skipped": 0,<br /> "failed": 0<br /> },<br /> "hits": {<br /> "total": 2,<br /> "max_score": 1,<br /> "hits": [<br /> {<br /> "_index": "kafka_es_test",<br /> "_type": "kafka-connect",<br /> "_id": "kafka_es_test+0+0",<br /> "_score": 1,<br /> "_source": {<br /> "nickname": "michel"<br /> }<br /> },<br /> {<br /> "_index": "kafka_es_test",<br /> "_type": "kafka-connect",<br /> "_id": "kafka_es_test+0+1",<br /> "_score": 1,<br /> "_source": {<br /> "nickname": "mushao"<br /> }<br /> }<br /> ]<br /> }<br /> }<br />

可以看到数据已经被成功写入

3 Confluent CLI

3.1 简介

查阅资料时发现很多文章都是使用Confluent CLI启动Kafka Connect，然而官方文档已经明确说明了该CLI只是适用于开发阶段，不能用于生产环境。

它可以一键启动包括zookeeper，kafka，schema registry, kafka rest, connect等在内的多个服务。但是这些服务对于Kafka Connect都不是必须的，如果不使用AvroConverter，则只需要启动Connect即可。即使使用了AvroConverter, 也只需要启动schema registry，将schema保存在远端的kafka中。Kafka Connect REST API也只是为用户提供一个管理connector的接口，也不是必选的。

另外使用CLI启动默认配置为启动Distributed的Connector，需要通过环境变量来修改配置

3.2 使用Confluent CLI

confluent CLI提供了丰富的命令，包括服务启动，服务停止，状态查询，日志查看等，详情参考如下简介视频 [[Introducing the Confluent CLI | Screencast](https://www.youtube.com/watch?v=ZKqBptBHZTg)]

1) 启动

<br /> ./bin/confluent start<br />

2) 检查confluent运行状态

<br /> ./bin/confluent status<br />

当得到如下结果则说明confluent启动成功

<br /> ksql-server is [UP]<br /> connect is [UP]<br /> kafka-rest is [UP]<br /> schema-registry is [UP]<br /> kafka is [UP]<br /> zookeeper is [UP]<br />

3) 问题定位

如果第二步出现问题，可以使用log命令查看，如connect未启动成功则

<br /> ./bin/confluent log connect<br />

4) 加载Elasticsearch Connector

a) 查看connector

<br /> ./bin/confluent list connectors<br />

结果

<br /> Bundled Predefined Connectors (edit configuration under etc/):<br /> elasticsearch-sink<br /> file-source<br /> file-sink<br /> jdbc-source<br /> jdbc-sink<br /> hdfs-sink<br /> s3-sink<br />

b) 加载Elasticsearch connector

<br /> ./bin/confluent load elasticsearch-sink<br />

结果

<br /> {<br /> "name": "elasticsearch-sink",<br /> "config": {<br /> "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",<br /> "tasks.max": "1",<br /> "topics": "kafka_es_test",<br /> "key.ignore": "true",<br /> "connection.url": "<a href="http://192.168.0.8:9200"" rel="nofollow" target="_blank">http://192.168.0.8:9200"</a>,<br /> "type.name": "kafka-connect",<br /> "name": "elasticsearch-sink"<br /> },<br /> "tasks": [],<br /> "type": null<br /> }<br />

5) 使用producer生产数据，并使用kibana验证是否写入成功

4 Kafka Connect Rest API

Kafka Connect提供了一套完成的管理Connector的接口，详情参考[[Kafka Connect REST Interface](https://docs.confluent.io/curr ... i.html)]。该接口可以实现对Connector的创建，销毁，修改，查询等操作

1) GET connectors 获取运行中的connector列表

2) POST connectors 使用指定的名称和配置创建connector

3) GET connectors/(string:name) 获取connector的详细信息

4) GET connectors/(string:name)/config 获取connector的配置

5) PUT connectors/(string:name)/config 设置connector的配置

6) GET connectors/(string:name)/status 获取connector状态

7) POST connectors/(stirng:name)/restart 重启connector

8) PUT connectors/(string:name)/pause 暂停connector

9) PUT connectors/(string:name)/resume 恢复connector

10) DELETE connectors/(string:name)/ 删除connector

11) GET connectors/(string:name)/tasks 获取connectors任务列表

12) GET /connectors/(string: name)/tasks/(int: taskid)/status 获取任务状态

13) POST /connectors/(string: name)/tasks/(int: taskid)/restart 重启任务

14) GET /connector-plugins/ 获取已安装插件列表

15) PUT /connector-plugins/(string: name)/config/validate 验证配置

5 总结

Kafka Connect是Kafka一个功能强大的组件，为kafka提供了与外部系统连接的一套完整方案，包括数据传输，连接管理，监控，多副本等。相对于Logstash Kafka插件，功能更为全面，但配置也相对为复杂些。有文章提到其性能也优于Logstash Kafka Input插件，如果对写入性能比较敏感的场景，可以在实际压测的基础上进行选择。另外由于直接将数据从Kafka写入Elasticsearch, 如果需要对文档进行处理时，选择Logstash可能更为方便。