搜索客，搜索人自己的社区

一次filebeat i/o timeout 问题记录-ES内存引起

Beats • ziyou 发表了文章 • 2 个评论 • 8415 次浏览 • 2019-01-07 16:17 • 来自相关话题

问题
kibana展示数据表明数据采集中断了，没有新的日志数据进来了。
排查
查看logstash日志：

[2019-01-07T14:59:27,435][INFO ][org.logstash.beats.BeatsHandler] Exception: Connection reset by peer

[2019-01-07T14:59:29,870][INFO ][org.logstash.beats.BeatsHandler] Exception: Connection reset by peer

[2019-01-07T14:59:29,870][INFO ][org.logstash.beats.BeatsHandler] Exception: Connection reset by peer

[2019-01-07T14:59:41,719][INFO ][org.logstash.beats.BeatsHandler] Exception: Connection reset by peer

[2019-01-07T14:59:42,777][INFO ][org.logstash.beats.BeatsHandler] Exception: Connection reset by peer

[2019-01-07T14:59:48,227][INFO ][org.logstash.beats.BeatsHandler] Exception: Connection reset by peer

查看filebeat日志：

2019-01-07T15:00:13+08:00 INFO No non-zero metrics in the last 30s

2019-01-07T15:00:43+08:00 INFO Non-zero metrics in the last 30s: libbeat.logstash.call_count.PublishEvents=1 libbeat.logstash.publish.write_bytes=241120

2019-01-07T15:00:48+08:00 ERR Failed to publish events (host: 10.68.24.138:5044:10200), caused by: read tcp 10.68.24.46:59310->10.68.24.138:5044: i/o timeout

2019-01-07T15:00:48+08:00 INFO Error publishing events (retrying): read tcp 10.68.24.46:59310->10.68.24.138:5044: i/o timeout

2019-01-07T15:01:13+08:00 INFO Non-zero metrics in the last 30s: libbeat.logstash.publish.read_errors=1 libbeat.logstash.published_but_not_acked_events=2034

查看的初步结果是，filebeat连不上logstash，logstash一直重置filebeat的连接，但是这两个机器是一点问题没有。

日志看过了，没有明显的问题，那就按部就班一步一步查吧

1、先来最基本的，查看elasticsearch、logstash、filebeat是否启动。
2、网络，网络环境是之前配置好的，一直没有变的，网络的可能性小一些，但是也是使用telnet测试一下各个端口是不是通的。
3、logstash故障，查看是不是因为logstash的未知故障，记录logstash的日志，然后重启logstash，看看重启logstash后是否解决问题了。
4、日志，查看日志是否是在更新，在5分钟以内是否在更新，因为是在运行的环境，日志一般不会断，所以我把这个检查放在了第四步。
5、查看ES的硬盘和内存。

GET /_cat/allocation?v

GET _cat/nodes?v

问题排查到第五步已经发现原因了：ES其中一台机器的内存满了。

原因始末
在部署这套ELK环境的时候，由于服务器提供方当时提供的两台ES机器的内存不一样，一台是8G的，一台是4G的，所以在使用的的时候，我配置的ES的堆内存一台是4G，一台是2G；ES集群就两台机器，也没配置数据节点和客户端节点，其实三台、四台我也都不配置的，集群太小再分开配置，就没有服务器了。
开始使用的时候是没有问题的，但是当日志达到一定量的时候，2G的那台机器堆内存耗光了，然后就出现了日志不能采集的i/o timeout问题。

经验
在使用ELK的过程中，以上的五种原因导致的filebeat日志采集异常，我都遇见过，其中容易忽略的就是ES的内存和硬盘是否已经满了，当ES集群中其中一台机器的堆内存和硬盘满了的话，都会引起日志采集异常。所以在配置ES集群的时候最好所有的data节点的内存和硬盘配置一致。

一、开发环境

1、组件版本

CDH 集群版本：6.0.1
Spark 版本：2.2.0
Kafka 版本：1.0.1
ES 版本：6.5.1

2、Maven 依赖

```xml

org.scala-lang
scala-library
2.11.8

org.apache.spark
spark-core_2.11
2.2.0

org.apache.spark
spark-streaming_2.11
2.2.0

org.apache.spark
spark-streaming-kafka-0-10_2.11
2.2.0

org.apache.zookeeper
zookeeper
3.4.5-cdh6.0.1

org.elasticsearch
elasticsearch-spark-20_2.11
6.5.4

commons-httpclient
commons-httpclient
3.1

```

### 3、注意事项
如果使用 CDH 版本的 Spark，则在调试及实际部署运行的时候会出现下面的错误：
```
java.lang.ClassNotFoundException: org.apache.commons.httpclient.protocol.Protocol
```

很显然是缺少 httpclient 相关依赖造成的，对比开源版本与 CDH 版本的 Spark，发现开源版本多出了 `commons-httpclient-3.1.jar`，因此上述 Maven 的 pom 文件添加上对其依赖即可。

## 二、ES-Hadoop
### 1、简介
ES-Hadoop 实现了 Hadoop 生态（Hive、Spark、Pig、Storm 等）与 ElasticSearch 之间的数据交互，借助该组件可以将 Hadoop 生态的数据写入到 ES 中，然后借助 ES 对数据快速进行搜索、过滤、聚合等分析，进一步可以通过 Kibana 来实现数据的可视化。

同时，也可以借助 ES 作为数据存储层（类似数仓的 Stage 层或者 ODS 层），然后借助 Hadoop 生态的数据处理工具（Hive、MR、Spark 等）将处理后的数据写入到 HDFS 中。

> 使用 ES 做为原始数据的存储层，可以很好的进行数据去重、数据质量分析，还可以提供一些即时的数据服务，例如趋势展示、汇总分析等。

![对 Hadoop 数据进行交互分析](http://img.luooqi.com/FvvdyhdXXZxgU1fwqLDWrZHvK6ZB)

### 2、组成
ES-Hadoop 是一个整合性质的组件，它封装了 Hadoop 生态的多种组件与 ES 交互的 API，如果你只需要部分功能，可以使用细分的组件：
elasticsearch-hadoop-mr
elasticsearch-hadoop-hive
elasticsearch-hadoop-pig
elasticsearch-spark-20_2.10
elasticsearch-hadoop-cascading
elasticsearch-storm

三、elasticsearch-spark

1、配置

es-hadoop 核心是通过 es 提供的 restful 接口来进行数据交互，下面是几个重要配置项，更多配置信息请参阅[官方说明](https://www.elastic.co/guide/e ... n.html)：
es.nodes：需要连接的 es 节点（不需要配置全部节点，默认会自动发现其他可用节点）；
es.port：节点 http 通讯端口；
es.nodes.discovery：默认为 true，表示自动发现集群可用节点；
es.nodes.wan.only：默认为 false，设置为 true 之后，会关闭节点的自动 discovery，只使用 es.nodes 声明的节点进行数据读写操作；如果你需要通过域名进行数据访问，则设置该选项为 true，否则请务必设置为 false；
es.index.auto.create：是否自动创建不存在的索引，默认为 true；
es.net.http.auth.user：Basic 认证的用户名；
es.net.http.auth.pass：Basic 认证的密码。

scala val conf = new SparkConf().setIfMissing("spark.app.name","rt-data-loader").setIfMissing("spark.master", "local[5]") conf.set(ConfigurationOptions.ES_NODES, esNodes) conf.set(ConfigurationOptions.ES_PORT, esPort) conf.set(ConfigurationOptions.ES_NODES_WAN_ONLY, "true") conf.set(ConfigurationOptions.ES_INDEX_AUTO_CREATE, "true") conf.set(ConfigurationOptions.ES_NODES_DISCOVERY, "false") conf.set(ConfigurationOptions.ES_NET_HTTP_AUTH_USER, esUser) conf.set(ConfigurationOptions.ES_NET_HTTP_AUTH_PASS, esPwd) conf.set("es.write.rest.error.handlers", "ignoreConflict") conf.set("es.write.rest.error.handler.ignoreConflict", "com.jointsky.bigdata.handler.IgnoreConflictsHandler") 

特别需要注意的配置项为 es.nodes.wan.only，由于在云服务器环境中，配置文件使用的一般为内网地址，而本地调试的时候一般使用外网地址，这样将 es.nodes 配置为外网地址后，最后会出现节点找不到的问题（由于会使用节点配置的内网地址去进行连接）：
 org.elasticsearch.hadoop.EsHadoopIllegalArgumentException: No data nodes with HTTP-enabled available; node discovery is disabled and none of nodes specified fit the criterion [xxx.xx.x.xx:9200] 

此时将 es.nodes.wan.only 设置为 true 即可。推荐开发测试时使用域名，集群部署的时候将该选项置为 false。

2、屏蔽写入冲突

如果数据存在重复，写入 ES 时往往会出现数据写入冲突的错误，此时有两种解决方法。

方法一：设置 es.write.operation 为 upsert，这样达到的效果为如果存在则更新，不存在则进行插入，该配置项默认值为 index。

方法二：自定义冲突处理类，类似上述配置中设置了自定义的 error.handlers，通过自定义类来处理相关错误，例如忽略冲突等：
java public class IgnoreConflictsHandler extends BulkWriteErrorHandler { public HandlerResult onError(BulkWriteFailure entry, DelayableErrorCollector<byte[]> collector) throws Exception { if (entry.getResponseCode() == 409) { StaticLog.warn("Encountered conflict response. Ignoring old data."); return HandlerResult.HANDLED; } return collector.pass("Not a conflict response code."); } } 

方法二可以屏蔽写入版本比预期的小之类的版本冲突问题。

3、RDD 写入 ES

EsSpark 提供了两种主要方法来实现数据写入：
saveToEs ：RDD 内容为 Seq[Map]，即一个 Map 对象集合，每个 Map 对应一个文档；
saveJsonToEs：RDD 内容为 Seq[String]，即一个 String 集合，每个 String 是一个 JSON 字符串，代表一条记录（对应 ES 的 _source）。

数据写入可以指定很多配置信息，例如：
es.resource：设置写入的索引和类型，索引和类型名均支持动态变量；
es.mapping.id：设置文档 _id 对应的字段名；
es.mapping.exclude：设置写入时忽略的字段，支持通配符。

scala val itemRdd = rdd.flatMap(line => { val topic = line.topic() println("正在处理：" + topic + " - " + line.partition() + " : " + line.offset()) val jsonArray = JSON.parseArray(line.value()).toJavaList(classOf[JSONObject]).asScala val resultMap = jsonArray.map(jsonObj =>{ var tmpId = "xxx" var tmpIndex = "xxxxxx" jsonObj.put("myTmpId", tmpId) jsonObj.put("myTmpIndex", tmpIndex) jsonObj.getInnerMap }) resultMap }) val mapConf = Map( ("es.resource" , "{myTmpIndex}/doc"), ("es.write.operation" , "upsert"), ("es.mapping.id" , "myTmpId"), ("es.mapping.exclude" , "myTmp*") ) EsSpark.saveToEs(itemRdd, mapConf) 

es.mapping.exclude 只支持 RDD 为 Map 集合（saveToEs），当为 Json 字符串集合时（saveJsonToEs）会提示不支持的错误信息；这个配置项非常有用，例如 myTmpId 作为文档 id，因此没有必要重复存储到 _source 里面了，可以配置到这个配置项，将其从 _source 中排除。

---
Any Code，Code Any！

扫码关注『AnyCode』，编程路上，一起前行。

![](https://user-gold-cdn.xitu.io/ ... D45449)