Elasticsearch是一个基于Lucene的搜索服务器，包含Windows、macOS、Linux版等。它提供了一个分布式多用户能力的全文搜索引擎，能很方便地使大量数据具有搜索、分析和探索的能力。本文介绍了如何在windows环境安装Elasticsearch、Kibana和Logstash，然后通过Metricbeat进行Logstash运行状态的监控，kiban的可视化查看，以及一些配置。

参考文章：分布式日志处理：ELK+Kafka实现日志收集

Elasticsearch

简介

Elasticsearch 是一个分布式、高扩展、高实时的搜索与数据分析引擎。它能很方便的使大量数据具有搜索、分析和探索的能力。充分利用Elasticsearch的水平伸缩性，能使数据在生产环境变得更有价值。

Elasticsearch是与名为Logstash的数据收集和日志解析引擎以及名为Kibana的分析和可视化平台一起开发的。这三个产品被设计成一个集成解决方案，称为Elastic Stack（以前称ELK stack）。

Elasticsearch可以用于搜索各种文档。它提供可扩展的搜索，具有接近实时的搜索，并支持多租户。Elasticsearch是分布式的，这意味着索引可以被分成分片，每个分片可以有0个或多个副本。每个节点托管一个或多个分片，并充当协调器将操作委托给正确的分片。再平衡和路由是自动完成的。相关数据通常存储在同一个索引中，该索引由一个或多个主分片和零个或多个复制分片组成。一旦创建了索引，就不能更改主分片的数量。

Elasticsearch使用Lucene，并试图通过JSON和Java API提供其所有特性。它支持facetting和percolating，如果新文档与注册查询匹配，这对于通知非常有用。另一个特性称为“网关”，处理索引的长期持久性；例如，在服务器崩溃的情况下，可以从网关恢复索引。Elasticsearch支持实时GET请求，适合作为NoSQL数据存储，但缺少分布式事务。

‌Lucene是Apache软件基金会维护的开源全文检索引擎工具包，用Java编写，提供高性能的索引和搜索功能，广泛应用于信息检索领域。

实现原理

首先用户将数据提交到Elasticsearch 数据库中，再通过分词控制器去将对应的语句分词，将其权重和分词结果一并存入数据，当用户搜索数据时候，再根据权重将结果排名，打分，再将返回结果呈现给用户。

Elasticsearch的主要特点

Elasticsearch是一个功能强大、易于使用和高性能的搜索和分析引擎，适用于各种大规模数据的搜索、聚合和分析需求。

  1. 分布式架构：数据可以分布在多个节点上。这样可以实现高可用性和负载均衡，同时也能通过水平扩展来处理大规模数据。
  2. 实时搜索和分析：Elasticsearch能够实时对大规模数据进行搜索和分析，搜索结果的响应时间通常在毫秒级别。它支持全文搜索、词条搜索、模糊搜索等多种搜索方式，并提供强大的查询语言。
  3. 多种数据类型支持：包括结构化数据、半结构化数据和非结构化数据。可以处理文本、数字、日期、地理位置等多种数据类型。
  4. 强大的全文搜索能力：Elasticsearch使用倒排索引来加速全文搜索，可以快速地找到包含关键词的文档。它支持分词、同义词、模糊匹配等功能，可以灵活地处理各种复杂的查询需求。
  5. 多种数据分析功能：包括聚合、过滤、排序、统计等。它可以对大规模数据进行复杂的统计和分析，帮助用户深入了解数据。
  6. 易于使用和集成：提供了友好的RESTful API，可以方便地与其他应用程序集成。它还有丰富的客户端库，支持Java、Python、.NET等多种编程语言。

相关概念

cluster（集群）

集群中有多个节点，其中有一个为主节点，这个主节点是可以通过选举产生的，主从节点是对于集群内部来说的。es的一个概念就是去中心化，字面上理解就是无中心节点，这是对于集群外部来说的，因为从外部来看es集群，在逻辑上是个整体，你与任何一个节点的通信和与整个es集群通信是等价的。

shards（索引分片）

es可以把一个完整的索引分成多个分片，这样的好处是可以把一个大的索引拆分成多个，分布到不同的节点上。构成分布式搜索。分片的数量只能在索引创建前指定，并且索引创建后不能更改。

replicas（索引副本）

es可以设置多个索引的副本，副本的作用：

  • 提高系统的容错性，当某个节点某个分片损坏或丢失时可以从副本中恢复。
  • 提高es的查询效率，es会自动对搜索请求进行负载均衡。

recovery（数据恢复或数据重新分布）

es在有节点加入或退出时会根据机器的负载对索引分片进行重新分配，挂掉的节点重新启动时也会进行数据恢复。

river（es的一个数据源）

代表es的一个数据源，也是其它存储方式（如：数据库）同步数据到es的一个方法。它是以插件方式存在的一个es服务，通过读取river中的数据并把它索引到es中，官方的river有couchDB的，RabbitMQ的，Twitter的，Wikipedia的。

gateway（索引快照的存储方式）

es默认是先把索引存放到内存中，当内存满了时再持久化到本地硬盘。gateway对索引快照进行存储，当这个es集群关闭再重新启动时就会从gateway中读取索引备份数据。es支持多种类型的gateway，有本地文件系统（默认），分布式文件系统，Hadoop的HDFS和amazon的s3云存储服务。

discovery.zen（自动发现节点机制）

es是一个基于p2p的系统，它先通过广播寻找存在的节点，再通过多播协议进行节点之间的通信，同时也支持点对点的交互。

Transport（内部节点或集群 与客户端的交互方式）

默认内部是使用tcp协议进行交互，同时它支持http协议（json格式）、thrift、servlet、memcached、zeroMQ等的传输协议（通过插件方式集成）。

windows下的安装

安装ElasticSearch：

①下载 elasticsearch-6.4.1.zip，下载链接：Download|Elasticsearch|Elastic

②直接解压至某目录，设置该目录为ES_HOME环境变量

③安装JDK，并设置JAVA_HOME环境变量

④在windows下，运行 %ES_HOME%\bin\elasticsearch.bat即可运行（双击运行）

启动成功后Elasticsearch会占用两个端口，9300是Elasticsearch集群节点之间通信的端口，9200是用户可以通过http（如浏览器）访问数据的入口。安装完成之后，可以在浏览器访问 http://127.0.0.1:9200/_cat/indices?v&pretty 查询es存储数据状态的命令。

安装head插件：

联网时，直接运行%ES_HOME%\bin\plugin --install mobz/elasticsearch-head。

不联网时，下载elasticsearch-head的zipball的master包，然后运行%ES_HOME%\bin\plugin --url file:///[path-to-downloadfile] --install head，其中[path-to-downloadfile]是下载后master包的绝对路径。

安装完成，重启服务，在浏览器打开 http://localhost:9200/_plugin/head/ 即可。

如果报错：http://localhost:9200 报错：received plaintext http traffic on an https channel, closing connection Netty4HttpChannel。原因是Elasticsearch在Windows下开启了安全认证，找到 config/ 目录下面的elasticsearch.yml配置文件，把安全认证开关从原先的 true 都改成 false ，实现免密登录访问即可：

1
2
3
xpack.security.enabled: false
xpack.security.http.ssl:
    enabled: false

Windows下ES内存配置修改

Windows环境下elasticsearch起来后占用内存过高，可以通过配置修改elasticsearch占用的内存大小。涉及到elasticsearch的安装目录config文件夹下面的两个配置文件。

  • 在elasticsearch.yml里面找到 # bootstrap.memory_lock: true，去掉#号，使之生效。
  • 在jvm.options里面，找到 ## -Xms4g和## -Xmx4g，把##去掉，将其启用。

Linux环境搭建Elasticsearch

  • elasticsearch-6.2.4.tar.gz 安装文件
  • elasticsearch-analysis-ik-6.2.4.zip 安装中文文件

安装 Elasticsearch

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
sudo sysctl -w vm.max_map_count=262144          #设置内存！每次启动都要给个运行内存！
tar -zxvf elasticsearch-6.2.4.tar.gz         	#指定目录下 解压
# 解压完成,设置elasticsearch-6.2.4/config/目录下：elasticsearch.yml文件保存这一些配置信息,端口9200...

# linunx 创建一个组 Elasticsearch 不能使用Root用户，需要由一个独立用户！
groupadd elsearch    #创建一个组
useradd elsearch -g elsearch -p elasticsearch #设置密码

# 解压elasticsearch中文安装包！
# unzip unzip 解压命令! zip压缩包解压,需要下载应用！ install 在线安装(需要网络！)！
yum install unzip    #下载unzip应用;
# 解压目录下文件！
unzip /usr/local/elasticsearch-analysis-ik-6.2.4.zip

# 授权用户操作elsearch  
chown -R elsearch:elsearch  elasticsearch-6.2.4
# 创建一个目录, 把中午解压包放进去。 就可以显示中文了！ 
mkdir -p /usr/local/elasticsearch-6.2.4/plugins/analysis-ik/    #elasticsearch-6.2.4/plugins下创建一个文件analysis-ik
# 中文解压文件,直接copy进去
cp -r /usr/local/elasticsearch/* /usr/local/elasticsearch-6.2.4/plugins/analysis-ik/

# Contos还需进行配置：修改 soft    nproc     4096
vi /etc/security/limits.d/20-nproc.conf        #修改 soft    nproc     4096 对应...

vi /etc/security/limits.conf                  #文件添加：↓↓
*               soft    nofile          65536
*               hard    nofile          65536
*               soft    nproc           4096
*               hard    nproc           4096
#授权：
chown -R elsearch:elsearch /usr/local/elasticsearch/

启动 Elasticsearch

1
2
3
4
5
6
7
8
9
10
# 切换用户
su elsearch
# 刷新
source /etc/profile
# 切换路径
cd /usr/local
cd elasticsearch-6.2.4/
cd bin
# elasticsearch  /bin目录下启动 
./elasticsearch                # 启动Elasticsearch

启动问题

启动莫名奇妙报错了： ERROR: [1] bootstrap checks failed [1]: max virtual memory areas vm.max_map_

  • 切换到root用户修改配置sysctl.conf 执行： vi /etc/sysctl.conf
  • 添加下面配置：vm.max_map_count=655360
  • 并执行命令：sysctl -p
  • 然后，重新启动elasticsearch，即可启动成功。

ElasticSearch实时推送数据

实时搜索

  1. 实时索引更新

    Elasticsearch 提供了实时索引更新功能，当文档被索引或更新时，它们立即可用于搜索。这意味着用户可以在提交数据后立即执行搜索，而无需等待索引的重建或更新。

  2. 近实时搜索

    Elasticsearch 还支持近实时搜索，这意味着文档在被索引后的短时间内就可以被搜索到。这通常需要毫秒级的延迟，使用户可以几乎即刻获得最新数据。

  3. 持续查询

    Elasticsearch 支持持续查询（continuous queries），这是一种持续监视数据更改并在变化发生时立即通知客户端的机制。这通过 Elasticsearch 的 Watcher 插件来实现，可用于实现实时监控和报警系统。

实时推送

  1. 基于查询的通知

    Elasticsearch 允许用户定义特定查询条件，并设置通知规则。当满足查询条件时，Elasticsearch 可以触发通知，将新数据或结果发送给用户。这可以通过 Elastic Stack 中的组件，如 Logstash 和 Kibana，来实现。

  2. 集成第三方通知服务

    Elasticsearch 可以与第三方通知服务集成（如邮件或短信服务），以通过这些服务向用户发送通知。这可以通过 Elasticsearch Watcher 插件来实现，它可以在数据满足条件时触发通知。

实时推送数据到前端

Elasticsearch实时推送数据到前端，通常涉及到Elasticsearch的实时数据搜索和更新功能，以及前端的技术栈（如使用JavaScript和WebSocket等技术）来实现实时数据的接收和展示。步骤和方案如下：

  1. 步骤 1: 启用Elasticsearch的实时功能

    1. 使用Elasticsearch的实时搜索功能： 当你索引或更新文档时，这些变更几乎可以立即反映在搜索结果中。

    2. 开启_refresh_interval： 你可以在索引设置中调整refresh_interval，以控制Elasticsearch索引的刷新频率。默认情况下，这个值是1秒，这意味着每秒Elasticsearch会刷新一次索引，确保最新的文档可以被搜索到。

      1
      2
      3
      4
      5
      6
      PUT /your_index/_settings
      {
        "index" : {
          "refresh_interval" : "1s"
        }
      }

  2. 步骤 2: 使用Elasticsearch的Watcher或XPack Alerting

    对于更复杂的实时通知需求，可用Elasticsearch的Watcher插件或XPack Alerting功能。它们允许基于某些条件触发动作，例如发送HTTP请求到你的前端服务器。

    设置Watcher： 创建一个Watcher来监控你的数据变化。例如，当某个字段的值达到特定条件时，触发一个动作。

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    PUT /_watcher/watch/my_watch
    {
      "trigger": {
        "schedule": {
          "interval": "10s"
        }
      },
      "input": {
        "search": {
          "request": {
            "indices": ["your_index"],
            "body": {
              "query": {
                "match_all": {}
              }
            }
          }
        }
      },
      "condition": {
        "compare": {
          "ctx.payload.hits.total.value": {
            "gt": 0
          }
        }
      },
      "actions": {
        "notify_frontend": {
          "http": {
            "method": "POST",
            "url": "http://your-frontend-server/api/update"
          }
        }
      }
    }

  3. 步骤 3: 前端实现实时数据接收

    1. 使用WebSocket： 可以在前端使用WebSocket来监听后端服务器的实时推送。后端可以使用如socket.io、websockets等库来实现WebSocket服务器。
    1
    2
    3
    4
    const socket = io('http://your-backend-server'); // 使用socket.io客户端连接后端服务器
    socket.on('data', (data) => { // 监听'data'事件，接收后端推送的实时数据
      console.log(data); // 处理接收到的数据，例如更新页面显示等
    });

    1. 轮询： 另一种简单的方法是使用轮询（定期从后端获取数据）。虽然这不是真正的实时推送，但在某些情况下可以作为临时解决方案。可以使用setInterval在 JavaScript 中实现：

      1
      2
      3
      4
      5
      6
      7
      setInterval(() => {
        fetch('http://your-backend-server/api/data') // 定期从后端获取数据
          .then(response => response.json())
          .then(data => {
            console.log(data); // 处理接收到的数据，例如更新页面显示等
          });
      }, 10000); // 每10秒获取一次数据

  4. 步骤 4: 测试和优化

    确保进行充分的测试，确保实时数据的准确性和性能。监控网络延迟和Elasticsearch的性能，根据需要进行调整。