[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

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

GET /_cat/allocation?v

GET _cat/nodes?v

本文将详细介绍利用 ES-Hadoop 将 Spark 处理的数据写入到 ES 中。

一、开发环境

1、组件版本

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

2、Maven 依赖

<!-- scala -->
<dependency>
  <groupId>org.scala-lang</groupId>
  <artifactId>scala-library</artifactId>
  <version>2.11.8</version>
</dependency>

<!-- spark 基础依赖 -->
<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-core_2.11</artifactId>
  <version>2.2.0</version>
</dependency>

<!-- spark-streaming 相关依赖 -->
<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-streaming_2.11</artifactId>
  <version>2.2.0</version>
</dependency>

<!-- spark-streaming-kafka 相关依赖 -->
<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
  <version>2.2.0</version>
</dependency>

<!-- zookeeper 相关依赖 -->
<dependency>
  <groupId>org.apache.zookeeper</groupId>
  <artifactId>zookeeper</artifactId>
  <version>3.4.5-cdh6.0.1</version>
</dependency>

<!-- Spark-ES 相关依赖 -->
<dependency>
    <groupId>org.elasticsearch</groupId>
    <artifactId>elasticsearch-spark-20_2.11</artifactId>
    <version>6.5.4</version>
</dependency>

<!-- Spark-ES 依赖的 HTTP 传输组件 -->
<dependency>
    <groupId>commons-httpclient</groupId>
    <artifactId>commons-httpclient</artifactId>
    <version>3.1</version>
</dependency>

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

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 接口来进行数据交互，下面是几个重要配置项，更多配置信息请参阅官方说明：

• 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 认证的密码。

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，通过自定义类来处理相关错误，例如忽略冲突等：

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：设置写入时忽略的字段，支持通配符。

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』，编程路上，一起前行。

1. 科普篇：ELK(ElasticSearch, Logstash, Kibana)搭建日志分析平台。 http://t.cn/EGMUSm2

2. 怎么用vue和es构建图书搜索应用(需翻墙）。 http://t.cn/EGMUO9j

3. 一周热点：罗振宇2018“时间的朋友”跨年演讲未删减全文。 http://t.cn/EGMU8Se

• 编辑：bsll

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

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

1. 400+节点的Elasticsearch集群运维经验。 http://t.cn/EbJXsTM

2. Elasticsearch index、create和update的源码分析。 http://t.cn/EbJSwbP

3. 全文搜索引擎选Elasticsearch还是Solr？。 http://t.cn/EbJSRp9

• 编辑：bsll

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

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

Elastic stack不停的迭代中狂奔。在最新的V6.5版本发布后，我们可以发现，其路线图已经越来越倾向于成为一个全栈的，全链路的，从下至上，从底层硬件资源数据一直到上层用户数据，从资源监控，到性能监控，直至安全审计的智能化运维系统了。这其中解决的一个痛点是：企业中存在各种各样的业务系统，每个系统又存在不同的数据源和存储数据的数据库；同时在运维管理层面上，又各种不同的监控系统（资源监控，性能监控，安全和审计）以及上层的可视化系统。这样，导致运维人员需要面对繁多的系统、入口和数据维度，在处理问题时，需要登录不同的平台，并且无法对数据进行有效的关联分析，因此，是迫切需要一个强大的平台，能够快速便捷的处理这些问题的。 我们可以看到，从不停发布的beats，以及beats里面不停添加的modules：

以及支持的各种数据指标模版：

elastic stack在加速将越来越多的数据需要汇聚到一起，并且提供一个统一的接口进入，这就是Kibana。这里，我不打算深入的比较Kibana和Grafana，简单说一句，就是grafana的主要场景只是dashboard，并不适合将其用作一个将所有数据集中到一起，需要做各种维度的查询，分析，安全检查等功能的portal。所以，这里我们讨论的是Kibana上如何展示其他数据源的数据。

为什么是prometheus而不是beats

在这个人人上云的时代，无论是open stack还是K8S，最流行的资源监控软件我看到的是prometheus，特别是以node_exporter为基础的一系列metric采集工具，为prometheus提供了各种维度的监控数据。而对应的，elastic stack也提供了类似filebeat, metricbeat, packatbeat等一系列工具和对应的数据template。

我没有深入使用过prometheus，但作为一个beats的资深用户，我还是感觉到beats还存在诸多的问题，特别是filebeat上幽灵般存在的内存占用过多的问题，导致大规模在所有的节点上安装beats成为一种风险。并且，最主要的一个点，我觉得是beats采集的数据，是依赖于整个elastic stack进行展示的，而作为云的运维人员，其关心的重点是每个虚拟机，容器的资源监控情况，metric和alart是其重心，而非query，search，security等功能。并且安装一个prometheus，比安装整个elastic stack简便得多，消耗的资源也小的多。所以，普遍的，从主机运维人员的角度，肯定首选安装prometheus的exporter来作资源的监控，而非安装beats。

为什么Kibana上需要集成prometheus的数据

正因为之前所讲的，我们试图使用elastic stack作为一个多维度的统一的数据入口和展示工具，要求我们必须能在Kibana看到硬件资源监控级别的指标，而elastic stack提供的beats等工具，却并不为云运维人员所待见（因为他们更喜欢prometheus，而非elastic stack，原因见以上）。因此，如果我们需要将elastic stack打造为一套全栈的智能运维系统，大多数情况下，prometheus成为必须跨越的一个槛。

将prometheus上的数据转移到elasticsearch

这是我想到的第一个解决方案。可惜logstash上没有prometheus的input plugins。所以，我们还是需要beats。注意，在metricbeat上有一个prometheus的module，号称可以 fetch metric from prometheus exporter，但实际上只是一个乌龙，这个module的并不能从成千上万的云主机或者容器中拉取数据，它能做的只是获取prometheus服务器节点prometheus这个进程的基本数据，然并卵。 这里给大家介绍的是两个社区提供prometheus相关的beats：

但建议大家还是自己写一个beat吧，代码可以参考prombeat。 不过如果你仔细观看prometheus里面的数据，都是num type的，将其转存到elasticsearch绝对不是一个经济的选择：

将grafana集成到kibana中

这是为什么我在一开始提到了grafana，虽然它不适合做portal，但是极其适合做dashboard，而kibana又是如此的开放，随便做个插件，可以轻松的跳转到grafana的dashboard上。而grafana与prometheus又是如此的登对，看看grafana上的各种专业而美丽的prometheus的dashboard吧：

我们要做的是做一个kibana的插件，然后将关键参数传递给grafana，并跳转：

虽然kibana和grafana是两个不同的工具，但并不妨碍我们将它们放在一起工作。而Kibana的开放性和基于插件的独立开发模式，让我们可以更方便的将各种好用的开源工具集成到一起，这里展示的Kibana与grafana和promethues的集成，希望能给到你一些微光。