1、在kibana里创建、管理、可视化汇总数据 http://t.cn/EqtKAf7

2、在Spring Boot使用ES教程 http://t.cn/Eqt9rC4

3、一周热点：GitHub提供免费私有Repo http://t.cn/EGmbUQX

• 编辑: bsll
• 归档：https://elasticsearch.cn/article/6328
• 订阅：https://tinyletter.com/elastic-daily

正如官方文档所自豪宣称的那样。Kibana更多的是一个平台，一个可以让插件独立开发，“独立部署”的可扩展性平台，可以让我们自由的发挥自己的想象力和能力，根据自己的需求往上添加原生Kibana所不提供的功能。你可以开发一个新的app，也可以只部署一个后台服务，也可以是一个隐藏的跳转页面，这些，都有赖于plugin的方式，自由的在kibana上install, update和remove。

问题描述

以上。看起来是比较美好的，但硬币的反面是kibana作为一个单页应用，任何都其他功能都是"/"路径下都一个子path，任何插件的安装（除非是一个纯后台的服务，但我没有测试）都需要和主页面、所有已安装都插件产生联系，即每次插件都变动，都需要将所有的页面和js重新bundle一次。这个捆绑不是简单的捆绑，而是经过优化后的打包操作，相当耗时。重点是，按照目前的方式，optimize(bundle)的过程必须是现场的，即必须在正在运行的kibana服务器上进行，因此在以下情况下你可能会遇到麻烦：

• 你的kibana服务作为一个生产服务，不能停
• 你没有给kibana做双活
• 因为只是一个前端，你给kibana分配的硬件资源很少（单核2G，双核4G）
• 你使用的是6.3之后的版本，kibana已经默认安装了xpack。或者你是之前的版本，自己手动安装了xpack

这时，你若是安装或者更新插件（包括remove插件），都可能会因为optimize过程占用大量的cpu和内存资源，而造成kibana停止服务响应。

这里有一个小tips，如果你开发了多个插件，需要同时更新当时候，安装当时候请使用命令kibana-plugin install --no-optimize file:///path_to_your_file，当全部的插件都安装完了之后，再重启kibana，一次性的执行optimize流程，或者通过bin/kibana --optimize命令触发

Kibana架构简述

如果我们的目标是让kibana零等待时间升级插件，找到解决方案的前提是我们能够了解Kibana的软件架构和部署方式。

首先，我们需要知道的是Kibana是一个基于node的web应用，前端后端都主要使用的javascript。web后端使用的hapi作为web服务器应用程序。并且node无需安装，已经包含在了kibana目录下。（node目录）

以下是kibana的目录，所有的插件都安装在plugins目录，而所有打包后的内容都放在optimize目录。

├── LICENSE.txt
├── NOTICE.txt
├── README.txt
├── bin
├── config
├── data
├── node
├── node_modules
├── optimize
├── package.json
├── plugins
├── src
└── webpackShims

plugins目录(这里，我有两个插件)：

.
├── kibana_auth_plugin
│   ├── index.js
│   ├── node_modules
│   ├── package.json
│   ├── public
│   ├── server
│   └── yarn.lock
└── system_portal
    ├── index.js
    ├── node_modules
    ├── package.json
    ├── public
    ├── server
    └── yarn.lock

每个插件都是类似的目录结构。public目录存放的是前端的页面和js，server目录存放的是后端的js。这里最终要的信息是，插件的开发其实也是一种 前后端分离的架构 。插件安装后后端主程序会调用server目录下的文件，而前端public目录下的文件会被压缩后打包到optimize目录，详见如下。

optimize目录：

├── bundles
│   ├── 176bcca991b07a6ec908fc4d36ac5ae0.svg
│   ├── 45c73723862c6fc5eb3d6961db2d71fb.eot
│   ├── 4b5a84aaf1c9485e060c503a0ff8cadb.woff2
│   ├── 69d89e51f62b6a582c311c35c0f778aa.svg
│   ├── 76a4f23c6be74fd309e0d0fd2c27a5de.svg
│   ├── 7c87870ab40d63cfb8870c1f183f9939.ttf
│   ├── apm.bundle.js
│   ├── apm.entry.js
│   ├── apm.style.css
│   ├── kibana-auth-plugin.bundle.js
│   ├── kibana-auth-plugin.entry.js
│   ├── kibana-auth-plugin.style.css
│   ├── canvas.bundle.js
│   ├── canvas.entry.js
│   ├── canvas.style.css
│   ├── cc17a3dbad9fc4557b4d5d47a38fcc56.svg
│   ├── commons.bundle.js
│   ├── commons.style.css
│   ├── dashboardViewer.bundle.js
│   ├── dashboardViewer.entry.js
│   ├── dashboardViewer.style.css
│   ├── dfb02f8f6d0cedc009ee5887cc68f1f3.woff
│   ├── fa0bbd682c66f1187d48f74b33b5bbd0.svg
│   ├── graph.bundle.js
│   ├── graph.entry.js
│   ├── graph.style.css
│   ├── infra.bundle.js
│   ├── infra.entry.js
│   ├── infra.style.css
│   ├── kibana.bundle.js
│   ├── kibana.entry.js
│   ├── kibana.style.css
│   ├── ml.bundle.js
│   ├── ml.entry.js
│   ├── ml.style.css
│   ├── monitoring.bundle.js
│   ├── monitoring.entry.js
│   ├── monitoring.style.css
│   ├── space_selector.bundle.js
│   ├── space_selector.entry.js
│   ├── space_selector.style.css
│   ├── src
│   ├── stateSessionStorageRedirect.bundle.js
│   ├── stateSessionStorageRedirect.entry.js
│   ├── stateSessionStorageRedirect.style.css
│   ├── status_page.bundle.js
│   ├── status_page.entry.js
│   ├── status_page.style.css
│   ├── system_portal.bundle.js
│   ├── system_portal.entry.js
│   ├── system_portal.style.css
│   ├── timelion.bundle.js
│   ├── timelion.entry.js
│   ├── timelion.style.css
│   ├── vendors.bundle.js
│   └── vendors.style.css

前端浏览器在访问"/"目录的时候会最先获取到kibana.*.js相关的文件。我们看一下 kibana.entry.js, 里面是包含了所有插件的信息的，即，每次插件的变动，这些文件也会跟着跟新

/**
 * Kibana entry file
 *
 * This is programmatically created and updated, do not modify
 *
 * context: ä
  "env": "production",
  "kbnVersion": "6.5.0",
  "buildNum": 18730,
  "plugins": Ä
    "apm",
    "apm_oss",
    "beats_management",
    "kibana_auth_plugin",
    "canvas",
    "cloud",
    "console",
    "console_extensions",
    "dashboard_mode",
    "elasticsearch",
    "graph",
    "grokdebugger",
    "index_management",
    "infra",
    "input_control_vis",
    "inspector_views",
    "kbn_doc_views",
    "kbn_vislib_vis_types",
    "kibana",
    "kuery_autocomplete",
    "license_management",
    "logstash",
    "markdown_vis",
    "metric_vis",
    "metrics",
    "ml",
    "monitoring",
    "notifications",
    "region_map",
    "reporting",
    "rollup",
    "searchprofiler",
    "spaces",
    "state_session_storage_redirect",
    "status_page",
    "system_portal",
    "table_vis",
    "tagcloud",
    "tile_map",
    "tilemap",
    "timelion",
    "vega",
    "watcher",
    "xpack_main"
  Å
å
 */

// import global polyfills before everything else
import 'babel-polyfill';
import 'custom-event-polyfill';
import 'whatwg-fetch';
import 'abortcontroller-polyfill';
import 'childnode-remove-polyfill';

import ä i18n å from 'Ékbn/i18n';
import ä CoreSystem å from '__kibanaCore__'

const injectedMetadata = JSON.parse(document.querySelector('kbn-injected-metadata').getAttribute('data'));
i18n.init(injectedMetadata.legacyMetadata.translations);

new CoreSystem(ä
  injectedMetadata,
  rootDomElement: document.body,
  requireLegacyFiles: () => ä
    require('plugins/kibana/kibana');
  å
å).start()

优化部署的方案（前后端分离的部署）

我们已经初步了解了kibana和kibana plugins的架构。那kibana插件的安装方案是怎么样的呢？ kibana为了简化我们的工作，只需要我们将打包好的源码丢给kibana，然后执行命令：kibana-plugin install file:///path_to_your_file，这样貌似省事，但也把所有的工作都丢给了kibana服务器去完成。 在kibana服务器性能不佳的情况下，这部分工作可能会造成服务中断。因此，我们要代替kibana服务器完成这部分工作，做一个前后端分离的部署。

后端部署

后端部署的速度是极快的，只需要把文件解压缩到具体目录就可以：

`kibana-plugin install --no-optimize file:///path_to_your_file`

这里特别要注意: --no-optimize参数是必须的，这时，插件的安装只是一个解压的过程，不会让kibana服务器去做繁重的optimize工作。

注意，执行这一步之后，不能重启kibana服务器，否则会自动做optimize

前端部署

这里说的前端，主要是指bundle之后的内容。在你的开发环境上，安装插件。当插件安装完成后，把bundles目录整体打包（bundles.zip）。将打包好之后的内容，上传到kibana服务器，删除旧的optimize/bundles目录，把打包好的bundles目录解压到optimize目录下

注意，这里开发环境上的kibana版本，和kibana安装的插件必须是和生产环境上是一致的，否则会造成无法启动或者自动重做optimize

重启kibana服务器

当以上两步完成之后，重启kibana service即可，你会发现，内容已经更新，但是不会触发任何的optimize过程。

参考示例

以下是该过程的一个ansible playbook供大家参考

---

- name: deploy bundles zip
  copy: src=bundles.zip dest={{kibana_home}}/optimize force=yes mode={{file_mask}}

- name: deploy system plugins zip
  copy: src=system_portal-0.0.0.zip dest={{kibana_home}}/ force=yes mode={{file_mask}}

- name: deploy auth zip
  copy: src=kibana_auth_plugin-6.5.0.zip dest={{kibana_home}}/ force=yes mode={{file_mask}}

- name: remove system plugin
  shell: "{{kibana_home}}/bin/kibana-plugin remove system_portal"
  ignore_errors: True

- name: remove auth plugin
  shell: "{{kibana_home}}/bin/kibana-plugin remove kibana_auth_plugin"
  ignore_errors: True

- name: install system plugin
  shell: "{{kibana_home}}/bin/kibana-plugin install --no-optimize file://{{kibana_home}}/system_portal-0.0.0.zip"
  register: install_state

- name: install auth plugin
  shell: "{{kibana_home}}/bin/kibana-plugin install --no-optimize file://{{kibana_home}}/kibana_auth_plugin-6.5.0.zip"
  register: install_state
#  failed_when: "'Extraction complete' in install_state.stdout_lines"

- name: delete old bundls
  file: dest={{kibana_home}}/optimize/bundles state=absent

- name: delete old bundls
  unarchive:
     src: "{{kibana_home}}/optimize/bundles.zip"
     dest: "{{kibana_home}}/optimize/"
     copy: no
     group: "kibana"
     owner: "kibana"
     mode: "{{file_mask}}"

- name: delete zip files
  file: dest={{kibana_home}}/optimize/bundles.zip state=absent

- name: restart kibana
  become: yes
  service: name={{kibana_init_script | basename}} state=restarted enabled=yes

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

一、Logstash

1、Logstash 性能调优主要参数

• pipeline.workers：设置启动多少个线程执行 fliter 和 output；当 input 的内容出现堆积而 CPU 使用率还比较充足时，可以考虑增加该参数的大小；
• pipeline.batch.size：设置单个工作线程在执行过滤器和输出之前收集的最大事件数，较大的批量大小通常更高效，但会增加内存开销。输出插件会将每个批处理作为一个输出单元。；例如，ES 输出会为收到的每个批次发出批量请求；调整 pipeline.batch.size 可调整发送到 ES 的批量请求（Bulk）的大小；
• pipeline.batch.delay：设置 Logstash 管道的延迟时间， 管道批处理延迟是 Logstash 在当前管道工作线程中接收事件后等待新消息的最长时间（以毫秒为单位）；简单来说，当 pipeline.batch.size 不满足时，会等待 pipeline.batch.delay 设置的时间，超时后便开始执行 filter 和 output 操作。

2、使用 Ruby Filter 根据现有字段计算一个新字段

filter {
    ruby {
           code => "event.set('kpi', ((event.get('a') + event.get('b'))/(event.get('c')+event.get('d'))).round(2))"
     }
}

3、logstash filter 如何判断字段是够为空或者 null

if ![updateTime]

4、Date Filter 设置多种日期格式

date {
  match => ["logtime", "yyyy-MM-dd HH:mm:ss.SSS","yyyy-MM-dd HH:mm:ss,SSS"]
  target => "logtime_utc"
}

二、Elasticsearch

1、高效翻页 Search After

通常情况下我们会使用 from 和 size 的方式实现查询结果的翻页，但是当达到深度分页时，成本变得过高（堆内存占用和时间耗费与 from+size 的大小成正比），因此 ES 设置了限制（index.max_result_window），默认值为 10000，防止用户进行过于深入的翻页。

推荐使用 Scroll api 进行高效深度滚动，但滚动上下文代价很高，因此不要将 Scroll 用于实时用户请求。search_after 参数通过提供实时游标来解决深度滚动的问题，其主要思路是使用上一页的结果来帮助检索下一页。

GET twitter/_search
{
    "size": 10,
    "query": {
        "match" : {
            "title" : "elasticsearch"
        }
    },
    "search_after": [1463538857, "654323"],
    "sort": [
        {"date": "asc"},
        {"tie_breaker_id": "asc"}
    ]
}

2、ES 文档相似度 BM25 参数设置

ES2.X 默认是以 TF/IDF 算法计算文档相似度，从 ES5.X 开始，BM25 作为默认的相似度计算算法。

PUT /index
{
    "settings" : {
        "index" : {
            "similarity" : {
              "my_similarity" : {
                "type" : "DFR",
                "basic_model" : "g",
                "after_effect" : "l",
                "normalization" : "h2",
                "normalization.h2.c" : "3.0"
              }
            }
        }
    }
}

PUT /index/_mapping/_doc
{
  "properties" : {
    "title" : { "type" : "text", "similarity" : "my_similarity" }
  }
}

3、ES2.X 得分计算

得分计算脚本：

double tf = Math.sqrt(doc.freq); 
double idf = Math.log((field.docCount+1.0)/(term.docFreq+1.0)) + 1.0; 
double norm = 1/Math.sqrt(doc.length); 
return query.boost * tf * idf * norm;

• 忽略词频统计及词频位置：将字段的 index_options 设置为 docs;
• 忽略字段长度：设置字段的 "norms": { "enabled": false }；

4、CircuitBreakingException: [parent] Data too large

报错信息：

[WARN ][r.suppressed             ] path: /, params: {}
org.elasticsearch.common.breaker.CircuitBreakingException: [parent] Data too large, data for [<http_request>] would be [1454565650/1.3gb], which is larger than the limit of [1454427340/1.3gb], usages [request=0/0b, fielddata=568/568b, in_flight_requests=0/0b, accounting=1454565082/1.3gb]

jvm 堆内存不够当前查询加载数据所以会报 data too large, 请求被熔断，indices.breaker.request.limit默认为 jvm heap 的 60%，因此可以通过调整 ES 的 Heap Size 来解决该问题。

5、ES 免费的自动化运维工具推荐

• Ansible: https://github.com/elastic/ansible-elasticsearch
• Puppet: https://github.com/elastic/puppet-elasticsearch
• Cookbook: https://github.com/elastic/cookbook-elasticsearch
• Curator：https://www.elastic.co/guide/en/elasticsearch/client/curator/current/about.html

6、elasticsearch-hanlp 分词插件包

核心功能：

• 内置多种分词模式，适合不同场景；
• 内置词典，无需额外配置即可使用；
• 支持外置词典，用户可自定义分词算法，基于词典或是模型；
• 支持分词器级别的自定义词典，便于用于多租户场景；
• 支持远程词典热更新（待开发）；
• 拼音过滤器、繁简体过滤器（待开发）；
• 基于词语或单字的 ngram 切分分词（待开发）。

https://github.com/AnyListen/elasticsearch-analysis-hanlp

7、节点重启时延迟索引分片重分配

当某个节点短时间离开集群时，一般是不会影响整体系统运行的，可以通过下面的请求延迟索引分片的再分配。

PUT _all/_settings
{
  "settings": {
    "index.unassigned.node_left.delayed_timeout": "5m"
  }
}

8、ES 数据修改后，查询还是未修改前的数据

默认是 1 秒可见，如果你的需求一定要写完就可见，那在写的时候增加 refresh 参数，强制刷新即可，但强烈建议不这么干，因为这样会把整个集群拖垮。

9、Terms Query 从另一个索引获取 terms

当 Terms Query 需要指定很多 terms 的时候，如果手动设置还是相当麻烦的，可以通过 terms-lookup 的方式从另外一个索引加载需要匹配的 terms。

PUT /users/_doc/2
{
    "followers" : ["1", "3"]
}

PUT /tweets/_doc/1
{
    "user" : "1"
}

GET /tweets/_search
{
    "query" : {
        "terms" : {
            "user" : {
                "index" : "users",
                "type" : "_doc",
                "id" : "2",
                "path" : "followers"
            }
        }
    }
}

-----------等效下面的语句--------------

PUT /users/_doc/2
{
 "followers" : [
   {
     "id" : "1"
   },
   {
     "id" : "2"
   }
 ]
}

10、ES 备份路径设置

报错信息：

doesn't match any of the locations specified by path.repo because this setting is empty

结局方案，修改 ES 的配置文件：

# 在 elasticsearch.yml 中添加下面配置来设置备份仓库路径 
path.repo: ["/home/test/backup/zty_logstash"]

11、Query cache 和 Filter cache 的区别

Filter cache 被重命名为 Node Query Cache，也就是说 Query cache 等同于 Filter cache；Query Cache 采用了 LRU 的缓存方式（当缓存满的时候，淘汰旧的不用的缓存数据），Query Cache 只缓存被用于 filter 上下文的内容。

12、Shard 大小需要考虑的因素有哪些？

Lucene 底层没有这个大小的限制，20-40GB 的这个区间范围本身就比较大，经验值有时候就是拍脑袋，不一定都好使。

• Elasticsearch 对数据的隔离和迁移是以分片为单位进行的，分片太大，会加大迁移成本；
• 一个分片就是一个 Lucene 的库，一个 Lucene 目录里面包含很多 Segment，每个 Segment 有文档数的上限，Segment 内部的文档 ID 目前使用的是 Java 的整型，也就是 2 的 31 次方，所以能够表示的总的文档数为 Integer.MAX_VALUE - 128 = 2^31 - 128 = 2147483647 - 1 = 2,147,483,519，也就是21.4亿条；
• 同样，如果你不 force merge 成一个 Segment，单个 shard 的文档数能超过这个数；
• 单个 Lucene 越大，索引会越大，查询的操作成本自然要越高，IO 压力越大，自然会影响查询体验；
• 具体一个分片多少数据合适，还是需要结合实际的业务数据和实际的查询来进行测试以进行评估。

13、ES 索引更新时通过 mapping 限制指定字段更新

Elasticsearch 默认是 Dynamic Mapping，新字段会自动猜测数据类型，并自动 merge 到之前的 Mapping，你可以在 Mapping 里面可以配置字段是否支持动态加入，设置参数dynamic即可：true，默认，表示支持动态加入新字段；false，表示忽略该字段的后续索引等操作，但是索引还是成功的；strict支持不支持未知字段，直接抛错。

14、ES 数据快照到 HDFS

ES 做快照和使用 ES-Hadoop 导数据是完全的两种不同的方式，使用 ES-Hadoopp 后期导入的成本可能也不小。

• 如果要恢复快，当然是做快照和还原的方式最快，速度完全取决于网络和磁盘的速度；
• 如果为了节省磁盘，快照的时候，可以选 6.5 最新支持的 source_only 模式，导出的快照要小很多，不过恢复的时候要进行重建，速度慢。

15、segment.memory 简介

segment 的大小，和 indexing buffer 有关，有三种方式会生成 segment：

• 一种是 indexing buffer 写满了会生成 segment 文件，默认是堆内存的10%，是节点共享的；
• 一种是 index buffer 有文档，但是还没满，但是 refresh 时间到了，这个时候就会把 buffer 里面的生成 segment 文件；
• 还有最后一种就是 es 自动的会将小的 segment 文件定期合并产生新的 segment 文件。

三、社区文章精选

• 2018 年 Elastic Advent Calendar 分享活动
• 使用 ES-Hadoop 将 Spark Streaming 流数据写入 ES
• Elastic Stack 6.5 最新功能
• 让Elasticsearch飞起来!——性能优化实践干货

Any Code，Code Any！

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

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