1.如何在Elasticsearch中查找和删除重复文档
http://t.cn/EUxkESo
2.Elasticsearch线程池分析
http://t.cn/EUxkDNg
3.Elasticsearch Pipeline Aggregations指南
http://t.cn/EUxFzZX

编辑：金桥
归档：https://elasticsearch.cn/article/6195
订阅：https://tinyletter.com/elastic-daily

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/d ... gt%3B 保存在/usr/local下
  
  
• 安装插件
  
  <br /> cd /usr/local/es/elasticsearch-5.0.2/bin<br /> ./elasticsearch-plugin install file:///usr/local/repository-hdfs-5.0.2.zip<br />
  
  
• 安装成功后需要重启Elasticsearch
  
  

  备份与恢复
  

  
  

• 构建一个仓库
  
  <br /> PUT <a href="http://192.168.10.74:9200/_snapshot/backup" rel="nofollow" target="_blank">http://192.168.10.74:9200/_snapshot/backup</a><br /> { <br /> "type": "hdfs", <br /> "settings": { <br /> "uri": "hdfs://192.168.10.170:9000", <br /> "path": "/es", <br /> "conf_location": "/usr/local/hadoop/etc/hadoop/hdfs-site.xml" <br /> }<br /> }<br />
  
  
• 备份快照
  
  <br /> PUT <a href="http://192.168.10.74:9200/_snapshot/backup/snapshot_users?wait_for_completion=true" rel="nofollow" target="_blank">http://192.168.10.74:9200/_sna ... Dtrue</a><br /> {<br /> "indices": "users", //备份users的index，注意不设置这个属性，默认是备份所有index<br /> "ignore_unavailable": true,<br /> "include_global_state": false<br /> }<br />
  
  
• 恢复快照
  
  <br /> POST <a href="http://192.168.10.74:9200/_snapshot/backup/snapshot_users/_restore" rel="nofollow" target="_blank">http://192.168.10.74:9200/_sna ... store</a><br /> {<br /> "indices": "users", //指定索引恢复，不指定就是所有<br /> "ignore_unavailable": true, //忽略恢复时异常索引<br /> "include_global_state": false //是否存储全局转态信息,fasle代表有一个或几个失败，不会导致整个任务失败<br /> }<br />
  
  
  

  整合Spark与Elasticsearch
  

  
  

  整体思路
  

  
  

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

  引入依赖
  

  
  ```java
  

  
  org.elasticsearch
  elasticsearch-hadoop
  5.0.2
  
  ```
  
  
  
  ### 具体流程
  
  
• 数据在HDFS上，数据存储在HDFS的每个DataNode的block上
  
  
  
  

  
  
  
  

• 数据加载到Spark SQL
  
  
  • 数据从HDFS加载到Spark SQL中，以RDD形式存储
    
    java<br /> JavaRDD<String> textFile = spark.read().textFile("hdfs://192.168.10.170:9000/csv/user.csv")<br />
    
    
  • 添加数据结构信息转换为新的RDD
    
    java<br /> JavaRDD<UserItem> dataSplits = textFile.map(line -> {<br /> String records = line.toString().trim();<br /> String record = records.substring(0,records.length() - 1).trim();<br /> String[] parts = record.split("\\|");<br /> UserItem u = new UserItem();<br /> u.setName(parts[0]);<br /> u.setAge(parts[1]);<br /> u.setHeight(parts[2]);<br /> return u;<br /> });<br />
    
    
  • 根据新的RDD创建DataFrame
    
    java<br /> DataSet<Row> ds = spark.createDataFrame(dataSplits, UserItem.class);<br />
    
    
    
    

    
    
    
    

  • 由Dataset创建索引，并写入ES
    
    <br /> JavaEsSparkSQL.saveToEs(ds, "es_spark/users");<br />
    
    
• 数据在ES中建立索引
  
  
  
  

  
  
  
  

• Spark SQL通过索引对ES中的数据进行查询
  
  java<br /> SparkSession spark = SparkSession.builder().appName("es-spark").master("local").config("es.index.auto.create", true).getOrCreate();<br /> Map<String, String> options = new HashMap<>();<br /> options.put("pushdown", "true");<br /> options.put("es.nodes","192.168.10.74:9200");<br /> <br /> Dataset<Row> df = spark.read().options(options).format("org.elasticsearch.spark.sql").load("es_spark/users");<br /> df.createOrReplaceTempView("users");<br /> <br /> Dataset<Row> userSet = spark.sql("SELECT name FORM users WHERE age >=10 AND age <= 20");<br /> userSet.show();<br />
  
  

  结束
  

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

1. 快来学习下如何便捷地记录文档插入 es 的时间？
http://t.cn/EUfESt7
2. 6.5.3发布了，还没升级到6的同学又多了一个新的选择！
http://t.cn/EUfEHvA
3. (自备梯子)如何不停机修改索引的 mapping？
http://t.cn/EUfErLr

编辑：rockybean
归档：https://elasticsearch.cn/article/6193
订阅：https://tinyletter.com/elastic-daily

为什么我的数据中没有updateTime 和 createTime 字段的；  理论上是不会执行if 里面的代码才对的；  但是为什么看日志输出好像是执行了if代码块的代码呢 
 
下面的是数据源， 并没有time字段的

{

  "仓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](https://www.elastic.co/guide/e ... k.html)，这里不展开介绍。
  
  
  

  3. 日志场景调优
  

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

  3.1 基础场景
  

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

• 建议打开最优压缩，一般可降低40%存储。
  
• 设置原始日志字段（message）为text，去除keyword类型子字段，提供全文搜索能力，降低存储。
  
• 关闭_all索引，前面已通过message提供全文搜索能力。
  
• 对于其他字符串字段，统一设置为keyword类型，避免默认情况下字符串字段同时存储text、keyword两种类型的数据。
  
• 使用开源组件（如Beats）上报数据时会包含较多辅助信息，用户可通过修改组件配置文件进行裁剪。
  
  这样去除message的keyword子字段、_all等冗余信息后，再加上最优压缩，可以保证数据相对精简。下面给出这类场景的常用模板，供用户参考：
  <br /> {<br /> "order": 5,<br /> "template": "my_log_*",<br /> "settings": {<br /> "translog.durability": "async",<br /> "translog.sync_interval": "5s",<br /> "index.refresh_interval": "30s",<br /> "index.codec": "best_compression" # 最优压缩<br /> },<br /> "mappings": {<br /> "_default_": {<br /> "_all": { # 关闭_all索引<br /> "enabled": false<br /> },<br /> "dynamic_templates": [<br /> {<br /> "log": { # 原始日志字段，分词建立索引<br /> "match": "message",<br /> "mapping": {<br /> "type": "text"<br /> }<br /> }<br /> },<br /> {<br /> "strings": { # 其他字符串字段，统一设置为keyword类型<br /> "match_mapping_type": "string",<br /> "mapping": {<br /> "type": "keyword"<br /> }<br /> }<br /> }<br /> ]<br /> }<br /> }<br /> }<br />
  
  
  

  3.2 精准搜索场景
  

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

• 清洗过程中，可仅解析出需要精准搜索的部分作为独立字段，用于精准搜索。
  
• 对于精准搜索字段，如果无排序/聚合需求，可以关闭doc_values；对于字符串，一般使用keyword，可按需考虑使用text。
  
  下面给出这类场景的常用模板，供用户参考：
  <br /> {<br /> "order": 5,<br /> "template": "my_log_*",<br /> "settings": {<br /> "translog.durability": "async",<br /> "translog.sync_interval": "5s",<br /> "index.refresh_interval": "30s",<br /> "index.codec": "best_compression" # 最优压缩<br /> },<br /> "mappings": {<br /> "_default_": {<br /> "_all": { # 关闭_all索引<br /> "enabled": false<br /> },<br /> "dynamic_templates": [<br /> {<br /> "log": { # 原始日志字段，分词建立索引<br /> "match": "message",<br /> "mapping": {<br /> "type": "text"<br /> }<br /> }<br /> },<br /> {<br /> "precise_fieldx": { # 精准搜索字段<br /> "match": "fieldx",<br /> "mapping": {<br /> "type": "keyword",<br /> "doc_values": false<br /> }<br /> }<br /> },<br /> {<br /> "strings": { # 其他字符串字段，统一设置为keyword类型<br /> "match_mapping_type": "string",<br /> "mapping": {<br /> "type": "keyword"<br /> }<br /> }<br /> }<br /> ]<br /> }<br /> }<br /> }<br />
  
  

  3.3 统计分析场景
  

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

• 清洗过程中，解析出所有需要的数据作为独立字段；原始日志非必要时，建议去除。
  
• 如果有强需求保留原始日志，可以设置该字段enabled属性为false，只存储不索引。
  
• 多数字段保持默认即可，会自动建立索引、打开doc_values，可用于查询、排序、聚合。
  
• 对部分无排序/聚合需求、开销高的字段，可以关闭doc_values。
  
  
  下面给出这类场景的常用模板，供用户参考：
  
  
  <br /> {<br /> "order": 5,<br /> "template": "my_log_*",<br /> "settings": {<br /> "translog.durability": "async",<br /> "translog.sync_interval": "5s",<br /> "index.refresh_interval": "30s",<br /> "index.codec": "best_compression" # 最优压缩<br /> },<br /> "mappings": {<br /> "_default_": {<br /> "_all": { # 关闭_all索引<br /> "enabled": false<br /> },<br /> "dynamic_templates": [<br /> {<br /> "log": { # 原始日志字段，关闭索引<br /> "match": "message",<br /> "mapping": {<br /> "enabled": false<br /> }<br /> }<br /> },<br /> {<br /> "index_only_fieldx": { # 仅索引的字段，无排序/聚合需求<br /> "match": "fieldx",<br /> "mapping": {<br /> "type": "keyword",<br /> "doc_values": false<br /> }<br /> }<br /> },<br /> {<br /> "strings": { # 其他字符串字段，统一设置为keyword类型<br /> "match_mapping_type": "string",<br /> "mapping": {<br /> "type": "keyword"<br /> }<br /> }<br /> }<br /> ]<br /> }<br /> }<br /> }<br />
  

  ES 5.1及之后的版本，支持关键字查询时[自动选择目标字段](https://www.elastic.co/guide/e ... _field)，用户没有必要再使用原始日志字段提供不指定字段进行查询的能力。
  
  
  

  4. 小结
  

  日志的使用方式比较灵活，本文结合常见的客户使用方式，从整体上对性能、成本进行优化。用户也可结合自身业务场景，参考文章[Elasticsearch调优实践](https://cloud.tencent.com/developer/article/1156231)进行更细致的优化。
  
  
  
  
  
  
  

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

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

安装和配置metricbeat

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

```

Module: kafka

Docs: https://www.elastic.co/guide/e ... .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的告警等功能了